ronantakizawa/codereview-bench huggingface dataset:
Repository: influxdata/influxdb
File: cmd/influxd/run/server.go (Go)

Code before:
	s.Subscriber.WithLogger(s.Logger)
	for _, svc := range s.Services {
		svc.WithLogger(s.Logger)
	}
	s.SnapshotterService.WithLogger(s.Logger)
	s.Monitor.WithLogger(s.Logger)

	// Open TSDB store.
	if err := s.TSDBStore.Open(); err != nil {
		return fmt.Errorf("open tsdb store: %s", err)
	}

	// Add the subscriber before opening the PointsWriter
	s.PointsWriter.Subscriber = s.Subscriber

	// Open the subscriber service
	if err := s.Subscriber.Open(); err != nil {
		return fmt.Errorf("open subscriber: %s", err)
	}

	// Open the points writer service
	if err := s.PointsWriter.Open(); err != nil {
		return fmt.Errorf("open points writer: %s", err)
	}

	s.Monitor.WithLimiter(s.TSDBStore.EngineOptions.CompactionThroughputLimiter)

	for _, service := range s.Services {
		if err := service.Open(); err != nil {
			return fmt.Errorf("open service: %s", err)
		}
	}

	// Start the reporting service, if not disabled.
	if !s.reportingDisabled {
		go s.startServerReporting()
	}

	return nil
}

// Close shuts down the meta and data stores and all services.
func (s *Server) Close() error {
	s.stopProfile()

	// Close the listener first to stop any new connections
	if s.Listener != nil {
		s.Listener.Close()
	}

	// Close services to allow any inflight requests to complete

Review comment (bug):
I think I'm going to change this to `WithCompactThroughputLimiter` or something. Thinking about if in the future we want to add other limiters to the metrics collection for /debug/vars. :thinking: 

Code after:
	s.Subscriber.WithLogger(s.Logger)
	for _, svc := range s.Services {
		svc.WithLogger(s.Logger)
	}
	s.SnapshotterService.WithLogger(s.Logger)
	s.Monitor.WithLogger(s.Logger)

	// Open TSDB store.
	if err := s.TSDBStore.Open(); err != nil {
		return fmt.Errorf("open tsdb store: %s", err)
	}

	// Add the subscriber before opening the PointsWriter
	s.PointsWriter.Subscriber = s.Subscriber

	// Open the subscriber service
	if err := s.Subscriber.Open(); err != nil {
		return fmt.Errorf("open subscriber: %s", err)
	}

	// Open the points writer service
	if err := s.PointsWriter.Open(); err != nil {
		return fmt.Errorf("open points writer: %s", err)
	}

	s.Monitor.WithCompactThroughputLimiter(s.TSDBStore.EngineOptions.CompactionThroughputLimiter)

	for _, service := range s.Services {
		if err := service.Open(); err != nil {
			return fmt.Errorf("open service: %s", err)
		}
	}

	// Start the reporting service, if not disabled.
	if !s.reportingDisabled {
		go s.startServerReporting()
	}

	return nil
}

// Close shuts down the meta and data stores and all services.
func (s *Server) Close() error {
	s.stopProfile()

	// Close the listener first to stop any new connections
	if s.Listener != nil {
		s.Listener.Close()
	}

	// Close services to allow any inflight requests to complete
Repository: influxdata/influxdb
File: monitor/stats.go (Go)

Code before:
	"github.com/influxdata/influxdb/pkg/limiter"
	"golang.org/x/time/rate"
)

// stats captures statistics
type stats struct {
	comp compactThroughputStats
}

type compactThroughputStats struct {
	limiter limiter.Rate
}

// CompactThroughputUsage calculates the percentage of burst capacity currently consumed by compaction.
func (s *stats) CompactThroughputUsage() float64 {
	percentage := 100 * (1 - rate.Limit(s.comp.limiter.Tokens())/s.comp.limiter.Limit())
	return float64(percentage)
}

func (s *stats) Diagnostics() (*diagnostics.Diagnostics, error) {
	compactThroughputUsage := s.CompactThroughputUsage()
	i := fmt.Sprintf("%.2f", compactThroughputUsage)
	compactThroughputUsageTrunc, err := strconv.ParseFloat(i, 2)
	if err != nil {
		return nil, err
	}

	d := map[string]interface{}{
		"compact-throughput-usage-percentage": compactThroughputUsageTrunc,
	}

	return diagnostics.RowFromMap(d), nil
}


Review comment (suggestion):
Rounding the the percentage to 2 decimal places is an excellent idea! The less significant digits of a floating point number are usually noise. This is not the best way to round it, though. Rounding to 2 decimal places would be:

```
compactThroughputUsageTrunc := math.Round(compactThroughputUsage * 100.0) / 100.0
```

This is less work and memory allocation than converting to a string and back. It also allows more control over how we round than using `fmt.Sprintf`. For instance, `math.Round` could be changed to `math.Ceil` or `math.Floor` if we wanted to change the rounding direction.

Code after:
	"golang.org/x/time/rate"
)

// stats captures statistics
type stats struct {
	comp compactThroughputStats
}

type compactThroughputStats struct {
	limiter limiter.Rate
}

// CompactThroughputUsage calculates the percentage of burst capacity currently consumed by compaction.
func (s *stats) CompactThroughputUsage() float64 {
	percentage := 100 * (1 - rate.Limit(s.comp.limiter.Tokens())/s.comp.limiter.Limit())
	return float64(percentage)
}

func (s *stats) Diagnostics() (*diagnostics.Diagnostics, error) {
	compactThroughputUsage := s.CompactThroughputUsage()
	compactThroughputUsageTrunc := math.Round(compactThroughputUsage*100.0) / 100.0
	d := map[string]interface{}{
		"compact-throughput-usage-percentage": compactThroughputUsageTrunc,
	}

	return diagnostics.RowFromMap(d), nil
}
Repository: influxdata/influxdb
File: tsdb/engine/tsm1/compact_test.go (Go)

Code before:
						Size: 500 * 1024 * 1024,
					},
					FirstBlockCount: 100,
				},
				{
					FileStat: tsm1.FileStat{
						Path: "05-03.tsm1",
						Size: 400 * 1024 * 1024,
					},
					FirstBlockCount: 10,
				},
			},
			testShardTime: -1,
			expectedResult: func() TestLevelResults {
				return TestLevelResults{
					level5Groups: []tsm1.PlannedCompactionGroup{
						{
							tsm1.CompactionGroup{
								"01-05.tsm1",
								"01-06.tsm1",
								"01-07.tsm1",
								"02-04.tsm1",
								"02-05.tsm1",
								"02-06.tsm1",
								"03-02.tsm1",
								"03-03.tsm1",
								"03-04.tsm1",
								"04-01.tsm1",
								"04-02.tsm1",
								"05-03.tsm1",
							}, tsdb.DefaultAggressiveMaxPointsPerBlock},
					},
				}
			},
		},
		{
			name: "1.12.0 RC0 Planner issue mock data from cluster",
			files: []tsm1.ExtFileStat{
				{
					FileStat: tsm1.FileStat{
						Path: "000029202-000000004.tsm",
						Size: 2254857830,
					},
				},
				{
					FileStat: tsm1.FileStat{
						Path: "000029202-000000005.tsm",
						Size: 2254857830,
					},
				},
				{

Review comment (refactor):
this is a duplicate tsm file in this test. there are two `03-03.tsm1` files in the input list. 

The question is what do we think the intent of the test is. Do we 
* delete the duplicate
* rename it as 05-03
* rename it as 04-03

I think the intent was that this should be the third file in the 04 generation as the 04 generation is a level 1 generation. By making this file 05-03... there's now a level 3 generation after the level 1 generation. I don't think that's the intent of _this test_ (although it is an interesting test case). 

please rename to 04-03

Code after:
						Size: 500 * 1024 * 1024,
					},
					FirstBlockCount: 100,
				},
				{
					FileStat: tsm1.FileStat{
						Path: "04-03.tsm1",
						Size: 400 * 1024 * 1024,
					},
					FirstBlockCount: 10,
				},
			},
			testShardTime: -1,
			expectedResult: func() TestLevelResults {
				return TestLevelResults{
					level5Groups: []tsm1.PlannedCompactionGroup{
						{
							tsm1.CompactionGroup{
								"01-05.tsm1",
								"01-06.tsm1",
								"01-07.tsm1",
								"02-04.tsm1",
								"02-05.tsm1",
								"02-06.tsm1",
								"03-02.tsm1",
								"03-03.tsm1",
								"03-04.tsm1",
								"04-01.tsm1",
								"04-02.tsm1",
								"04-03.tsm1",
							}, tsdb.DefaultAggressiveMaxPointsPerBlock},
					},
				}
			},
		},
		{
			name: "1.12.0 RC0 Planner issue mock data from cluster",
			files: []tsm1.ExtFileStat{
				{
					FileStat: tsm1.FileStat{
						Path: "000029202-000000004.tsm",
						Size: 2254857830,
					},
				},
				{
					FileStat: tsm1.FileStat{
						Path: "000029202-000000005.tsm",
						Size: 2254857830,
					},
				},
				{
Repository: influxdata/influxdb
File: tsdb/engine/tsm1/reader.go (Go)

Code before:
}

// InUse returns whether the TSMReader currently has any active references.
func (t *TSMReader) InUse() bool {
	refs := atomic.LoadInt64(&t.refs)
	return refs > 0
}

// Remove removes any underlying files stored on disk for this reader.
func (t *TSMReader) Remove() error {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.remove()
}

// Rename renames the underlying file to the new path.
func (t *TSMReader) Rename(path string) error {
	t.mu.Lock()
	defer t.mu.Unlock()
	var err error
	if err = t.accessor.rename(path); err != nil {
		return fmt.Errorf("failure renaming to %q: %w", path, err)
	}
	if t.parseFileNameFunc != nil {
		if t.generation, t.sequence, err = t.parseFileNameFunc(path); err != nil {
			return fmt.Errorf("failed parsing filename %q for generation and sequence numbers: %w", path, err)
		}
	}
	return nil
}

// Remove removes any underlying files stored on disk for this reader.
func (t *TSMReader) remove() error {
	path := t.accessor.path()

	if t.InUse() {
		return ErrFileInUse
	}

	if path != "" {
		err := os.RemoveAll(path)
		if err != nil {
			return err
		}
	}

	if err := t.tombstoner.Delete(); err != nil {
		return err
	}
	return nil
}

Review comment (suggestion):
This error adds context that the same error in `NewTSMReader` does not. I think the context here is redundant because the error from `DefaultParseFileName` (the only one we use) includes the filename.

I would remove the name from this context, but add that the parsing error occurred during `TSMReader.Rename`.

Code after:
	refs := atomic.LoadInt64(&t.refs)
	return refs > 0
}

// Remove removes any underlying files stored on disk for this reader.
func (t *TSMReader) Remove() error {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.remove()
}

// Rename renames the underlying file to the new path.
func (t *TSMReader) Rename(path string) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if err := t.accessor.rename(path); err != nil {
		return fmt.Errorf("failure renaming to %q: %w", path, err)
	}

	if err := t.parseAndCacheFileName(path); err != nil {
		return fmt.Errorf("rename failed: %w", err)
	}
	return nil
}

func (t *TSMReader) parseAndCacheFileName(path string) error {
	if t.parseFileNameFunc != nil {
		var err error
		t.generation, t.sequence, err = t.parseFileNameFunc(path)
		return err
	} else {
		// If parseFileNameFunc is nil, we are in a test or another TSMReader use
		// that does not involve compaction planning, and was not created by the FileStore
		return nil
	}
}

// Remove removes any underlying files stored on disk for this reader.
func (t *TSMReader) remove() error {
	path := t.accessor.path()

	if t.InUse() {
		return ErrFileInUse
	}

	if path != "" {
		err := os.RemoveAll(path)
		if err != nil {
			return err
		}
Repository: influxdata/influxdb
File: tsdb/engine/tsm1/reader.go (Go)

Code before:
	t.lastModified = stat.ModTime().UnixNano()
	if t.accessor == nil {
		t.accessor = &mmapAccessor{
			f:            f,
			mmapWillNeed: t.madviseWillNeed,
		}
	}

	index, err := t.accessor.init()
	if err != nil {
		return nil, err
	}

	t.index = index
	t.tombstoner = NewTombstoner(t.Path(), index.ContainsKey)

	if err := t.applyTombstones(); err != nil {
		return nil, err
	}

	if nil != t.parseFileNameFunc {
		// If parseFileNameFunc is nil, we are in a test or another TSMReader use
		// that does not involve compaction planning, and was not created by the FileStore
		t.generation, t.sequence, err = t.parseFileNameFunc(t.Path())
		if err != nil {
			return nil, err
		}
	}

	return t, nil
}

// WithObserver sets the observer for the TSM reader.
func (t *TSMReader) WithObserver(obs tsdb.FileStoreObserver) {
	t.tombstoner.WithObserver(obs)
}

func (t *TSMReader) applyTombstones() error {
	var cur, prev Tombstone
	batch := make([][]byte, 0, 4096)

	if err := t.tombstoner.Walk(func(ts Tombstone) error {
		cur = ts
		if len(batch) > 0 {
			if prev.Min != cur.Min || prev.Max != cur.Max {
				t.index.DeleteRange(batch, prev.Min, prev.Max)
				batch = batch[:0]
			}
		}

		// Copy the tombstone key and re-use the buffers to avoid allocations

Review comment (suggestion):
It might be a good idea to add context that the parse filename error occurred during the `NewTSMReader` call.

Code after:
	t.lastModified = stat.ModTime().UnixNano()
	if t.accessor == nil {
		t.accessor = &mmapAccessor{
			f:            f,
			mmapWillNeed: t.madviseWillNeed,
		}
	}

	index, err := t.accessor.init()
	if err != nil {
		return nil, err
	}

	t.index = index
	t.tombstoner = NewTombstoner(t.Path(), index.ContainsKey)

	if err := t.applyTombstones(); err != nil {
		return nil, err
	}

	err = t.parseAndCacheFileName(t.Path())
	if nil != err {
		return nil, fmt.Errorf("failed creating new TSM reader: %w", err)
	}

	return t, nil
}

// WithObserver sets the observer for the TSM reader.
func (t *TSMReader) WithObserver(obs tsdb.FileStoreObserver) {
	t.tombstoner.WithObserver(obs)
}

func (t *TSMReader) applyTombstones() error {
	var cur, prev Tombstone
	batch := make([][]byte, 0, 4096)

	if err := t.tombstoner.Walk(func(ts Tombstone) error {
		cur = ts
		if len(batch) > 0 {
			if prev.Min != cur.Min || prev.Max != cur.Max {
				t.index.DeleteRange(batch, prev.Min, prev.Max)
				batch = batch[:0]
			}
		}

		// Copy the tombstone key and re-use the buffers to avoid allocations
		n := len(batch)
		batch = batch[:n+1]
		if cap(batch[n]) < len(ts.Key) {
			batch[n] = make([]byte, len(ts.Key))
Repository: influxdata/influxdb
File: tsdb/engine/tsm1/reader.go (Go)

Code before:
		t.accessor = &mmapAccessor{
			f:            f,
			mmapWillNeed: t.madviseWillNeed,
		}
	}

	index, err := t.accessor.init()
	if err != nil {
		return nil, err
	}

	t.index = index
	t.tombstoner = NewTombstoner(t.Path(), index.ContainsKey)

	if err := t.applyTombstones(); err != nil {
		return nil, err
	}

	if nil != t.parseFileNameFunc {
		// If parseFileNameFunc is nil, we are in a test or another TSMReader use
		// that does not involve compaction planning, and was not created by the FileStore
		t.generation, t.sequence, err = t.parseFileNameFunc(t.Path())
		if err != nil {
			return nil, err
		}
	}

	return t, nil
}

// WithObserver sets the observer for the TSM reader.
func (t *TSMReader) WithObserver(obs tsdb.FileStoreObserver) {
	t.tombstoner.WithObserver(obs)
}

func (t *TSMReader) applyTombstones() error {
	var cur, prev Tombstone
	batch := make([][]byte, 0, 4096)

	if err := t.tombstoner.Walk(func(ts Tombstone) error {
		cur = ts
		if len(batch) > 0 {
			if prev.Min != cur.Min || prev.Max != cur.Max {
				t.index.DeleteRange(batch, prev.Min, prev.Max)
				batch = batch[:0]
			}
		}

		// Copy the tombstone key and re-use the buffers to avoid allocations
		n := len(batch)
		batch = batch[:n+1]

Review comment (refactor):
We may want to refactor this chunk of code into it's own method, something like `TSMReader.parseFileName`.  We repeat the same basic code twice and it might be nice to have a "safe" wrapper around `parseFileNameFunc`.

Code after:
		t.accessor = &mmapAccessor{
			f:            f,
			mmapWillNeed: t.madviseWillNeed,
		}
	}

	index, err := t.accessor.init()
	if err != nil {
		return nil, err
	}

	t.index = index
	t.tombstoner = NewTombstoner(t.Path(), index.ContainsKey)

	if err := t.applyTombstones(); err != nil {
		return nil, err
	}

	err = t.parseAndCacheFileName(t.Path())
	if nil != err {
		return nil, fmt.Errorf("failed creating new TSM reader: %w", err)
	}

	return t, nil
}

// WithObserver sets the observer for the TSM reader.
func (t *TSMReader) WithObserver(obs tsdb.FileStoreObserver) {
	t.tombstoner.WithObserver(obs)
}

func (t *TSMReader) applyTombstones() error {
	var cur, prev Tombstone
	batch := make([][]byte, 0, 4096)

	if err := t.tombstoner.Walk(func(ts Tombstone) error {
		cur = ts
		if len(batch) > 0 {
			if prev.Min != cur.Min || prev.Max != cur.Max {
				t.index.DeleteRange(batch, prev.Min, prev.Max)
				batch = batch[:0]
			}
		}

		// Copy the tombstone key and re-use the buffers to avoid allocations
		n := len(batch)
		batch = batch[:n+1]
		if cap(batch[n]) < len(ts.Key) {
			batch[n] = make([]byte, len(ts.Key))
		} else {
			batch[n] = batch[n][:len(ts.Key)]
Repository: influxdata/influxdb
File: influxdb3_py_api/src/lib.rs (Rust)

Code before:
            table_chunks: SerdeVecMap::<TableId, TableChunks>::new(),
            min_time_ns: 0,
            max_time_ns: 0,
        };

        let time_provider: Arc<dyn TimeProvider> =
            Arc::new(MockProvider::new(Time::from_timestamp_nanos(0)));
        let cache_store = Arc::new(Mutex::new(CacheStore::new(
            Arc::clone(&time_provider),
            Duration::from_secs(60),
        )));
        let py_cache = PyCache::new_test_cache(cache_store, "method_visibility".to_string());

        let query_executor: Arc<dyn QueryExecutor> = Arc::new(UnimplementedQueryExecutor);

        let plugin = r#"
def process_writes(influxdb3_local, table_batches, args=None):
    allowed = {"info", "warn", "error", "query", "write", "cache", "write_to_db"}
    attrs = {name for name in dir(influxdb3_local) if not name.startswith("__")}
    extras = attrs - allowed
    missing = allowed - attrs
    if extras or missing:
        raise RuntimeError(f"unexpected attributes: extras={sorted(extras)}, missing={sorted(missing)}")
"#;

        let result = system_py::execute_python_with_batch(
            plugin,
            &write_batch,
            schema,
            query_executor,
            None,
            None,
            &None::<HashMap<String, String>>,
            py_cache,
            None,
        );

        assert!(
            result.is_ok(),
            "PyPluginCallApi exposes unexpected Python methods: {result:?}"
        );
    }
}


Review comment (performance):
if you switch to `execute_schedule_trigger`, you'll get the same effect but won't need to construct the empty write_batch. 

Code after:
#[derive(Debug, thiserror::Error)]
pub enum ExecutePluginError {
    #[error(
        "the process_writes function is not present in the plugin. Should be defined as: process_writes(influxdb3_local, table_batches, args=None)"
    )]
    MissingProcessWritesFunction,

    #[error(
        "the process_request function is not present in the plugin. Should be defined as: process_request(influxdb3_local, query_parameters, request_headers, request_body, args=None) -> Tuple[str, Optional[Dict[str, str]]]"
    )]
    MissingProcessRequestFunction,

    #[error(
        "the process_scheduled_call function is not present in the plugin. Should be defined as: process_scheduled_call(influxdb3_local, call_time, args=None)"
    )]
    MissingProcessScheduledCallFunction,

    #[error("{0}")]
    PluginError(#[from] anyhow::Error),
}

pub mod logging;
pub mod system_py;
Repository: influxdata/influxdb
File: tsdb/store.go (Go)

Code before:
							zap.String("key", string(keyBuf)),
							zap.Error(tmpErr))
						errs = append(errs, tmpErr)
					}
				})
				if len(errs) != 0 {
					return errors.Join(errs...)
				}
			}

			const DeleteLogTrigger = 10_000
			seriesCount := ss.Cardinality()
			deleteStart := time.Now()
			var deletedCount atomic.Uint64
			var deleteLogCounter atomic.Uint64

			ss.ForEach(func(id uint64) {
				if err := sfile.DeleteSeriesID(id); err != nil {
					sfile.Logger.Error(
						"cannot delete series in shard",
						zap.Uint64("series_id", id),
						zap.Uint64("shard_id", shardID),
						zap.Error(err))
				}
				deleted := deletedCount.Add(1)
				deleteLogCounter.Add(1)

				if deleteLogCounter.Load() > DeleteLogTrigger {
					deleteLogCounter.Store(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)))
				}
			})
		}
	}

	// 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 (suggestion):
You can eliminate `deleteLogCounter` and use modulo of `deletedCount`. 

Code after:
							zap.String("key", string(keyBuf)),
							zap.Error(tmpErr))
						errs = append(errs, tmpErr)
					}
				})
				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))
			}
		}
	}