db: remove unnecessary metric updates

When we apply a new version, we also pass "delta metrics" that are
added to the current `LevelMetrics`.

Some of `LevelMetrics` fields (like table counts/sizes, estimated
ref size) are reset from the `LevelMetadata` from the new version.
These fields don't require maintenance through the delta metrics.

This change removes the ineffectual code which attempts to keep these
fields updated through the delta metrics.

Author: RaduBerinde

compaction.go

@@ -1520,22 +1520,6 @@ func (d *DB) runIngestFlush(c *tableCompaction) (*manifest.VersionEdit, error) {
 	var ingestSplitFiles []ingestSplitFile
 	ingestFlushable := c.flush.flushables[0].flushable.(*ingestedFlushable)
 
-	updateLevelMetricsOnExcise := func(m *manifest.TableMetadata, level int, added []manifest.NewTableEntry) {
-		levelMetrics := c.metrics.perLevel[level]
-		if levelMetrics == nil {
-			levelMetrics = &LevelMetrics{}
-			c.metrics.perLevel[level] = levelMetrics
-		}
-		levelMetrics.TablesCount--
-		levelMetrics.TablesSize -= int64(m.Size)
-		levelMetrics.EstimatedReferencesSize -= m.EstimatedReferenceSize()
-		for i := range added {
-			levelMetrics.TablesCount++
-			levelMetrics.TablesSize += int64(added[i].Meta.Size)
-			levelMetrics.EstimatedReferencesSize += added[i].Meta.EstimatedReferenceSize()
-		}
-	}
-
 	suggestSplit := d.opts.Experimental.IngestSplit != nil && d.opts.Experimental.IngestSplit() &&
 		d.FormatMajorVersion() >= FormatVirtualSSTables
 
@@ -1609,13 +1593,12 @@ func (d *DB) runIngestFlush(c *tableCompaction) (*manifest.VersionEdit, error) {
 				}
 				newFiles := applyExciseToVersionEdit(ve, m, leftTable, rightTable, layer.Level())
 				replacedTables[m.TableNum] = newFiles
-				updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
 			}
 		}
 	}
 
 	if len(ingestSplitFiles) > 0 {
-		if err := d.ingestSplit(context.TODO(), ve, updateLevelMetricsOnExcise, ingestSplitFiles, replacedTables); err != nil {
+		if err := d.ingestSplit(context.TODO(), ve, ingestSplitFiles, replacedTables); err != nil {
 			return nil, err
 		}
 	}
@@ -3615,9 +3598,6 @@ func (c *tableCompaction) makeVersionEdit(result compact.Result) (*manifest.Vers
 			outputMetrics.TablesFlushed++
 			outputMetrics.TableBytesFlushed += fileMeta.Size
 		}
-		outputMetrics.EstimatedReferencesSize += fileMeta.EstimatedReferenceSize()
-		outputMetrics.TablesSize += int64(fileMeta.Size)
-		outputMetrics.TablesCount++
 		outputMetrics.Additional.BytesWrittenDataBlocks += t.WriterMeta.Properties.DataSize
 		outputMetrics.Additional.BytesWrittenValueBlocks += t.WriterMeta.Properties.ValueBlocksSize
 	}

data_test.go

@@ -1190,7 +1190,6 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 		return versionUpdate{
 			VE:                      ve,
 			JobID:                   jobID,
-			Metrics:                 newFileMetrics(ve.NewTables),
 			InProgressCompactionsFn: func() []compactionInfo { return nil },
 		}, nil
 	})

file_cache_test.go

@@ -393,27 +393,11 @@ func TestVirtualReadsWiring(t *testing.T) {
 	d.checkVirtualBounds(v2)
 
 	// Write the version edit.
-	fileMetrics := func(ve *manifest.VersionEdit) levelMetricsDelta {
-		metrics := newFileMetrics(ve.NewTables)
-		for de, f := range ve.DeletedTables {
-			lm := metrics[de.Level]
-			if lm == nil {
-				lm = &LevelMetrics{}
-				metrics[de.Level] = lm
-			}
-			metrics[de.Level].TablesCount--
-			metrics[de.Level].TablesSize -= int64(f.Size)
-			metrics[de.Level].EstimatedReferencesSize -= f.EstimatedReferenceSize()
-		}
-		return metrics
-	}
-
 	applyVE := func(ve *manifest.VersionEdit) error {
 		_, err := d.mu.versions.UpdateVersionLocked(func() (versionUpdate, error) {
 			return versionUpdate{
 				VE:                      ve,
 				JobID:                   d.newJobIDLocked(),
-				Metrics:                 fileMetrics(ve),
 				InProgressCompactionsFn: func() []compactionInfo { return d.getInProgressCompactionInfoLocked(nil) },
 			}, nil
 		})

ingest.go

@@ -1869,7 +1869,6 @@ type ingestSplitFile struct {
 func (d *DB) ingestSplit(
 	ctx context.Context,
 	ve *manifest.VersionEdit,
-	updateMetrics func(*manifest.TableMetadata, int, []manifest.NewTableEntry),
 	files []ingestSplitFile,
 	replacedTables map[base.TableNum][]manifest.NewTableEntry,
 ) error {
@@ -1946,7 +1945,6 @@ func (d *DB) ingestSplit(
 				panic("ingest-time split produced a file that overlaps with ingested file")
 			}
 		}
-		updateMetrics(splitFile, s.level, added)
 	}
 	// Flatten the version edit by removing any entries from ve.NewFiles that
 	// are also in ve.DeletedFiles.
@@ -2113,14 +2111,7 @@ func (d *DB) ingestApply(
 					f.Level, sharedLevelsStart))
 			}
 			f.Meta = m
-			levelMetrics := metrics[f.Level]
-			if levelMetrics == nil {
-				levelMetrics = &LevelMetrics{}
-				metrics[f.Level] = levelMetrics
-			}
-			levelMetrics.TablesCount++
-			levelMetrics.TablesSize += int64(m.Size)
-			levelMetrics.EstimatedReferencesSize += m.EstimatedReferenceSize()
+			levelMetrics := metrics.level(f.Level)
 			levelMetrics.TableBytesIngested += m.Size
 			levelMetrics.TablesIngested++
 		}
@@ -2131,21 +2122,6 @@ func (d *DB) ingestApply(
 		// newer fileMetadata due to a split induced by another ingestion file, or an
 		// excise.
 		replacedTables := make(map[base.TableNum][]manifest.NewTableEntry)
-		updateLevelMetricsOnExcise := func(m *manifest.TableMetadata, level int, added []manifest.NewTableEntry) {
-			levelMetrics := metrics[level]
-			if levelMetrics == nil {
-				levelMetrics = &LevelMetrics{}
-				metrics[level] = levelMetrics
-			}
-			levelMetrics.TablesCount--
-			levelMetrics.TablesSize -= int64(m.Size)
-			levelMetrics.EstimatedReferencesSize -= m.EstimatedReferenceSize()
-			for i := range added {
-				levelMetrics.TablesCount++
-				levelMetrics.TablesSize += int64(added[i].Meta.Size)
-				levelMetrics.EstimatedReferencesSize += added[i].Meta.EstimatedReferenceSize()
-			}
-		}
 		var exciseBounds base.UserKeyBounds
 		if exciseSpan.Valid() {
 			exciseBounds = exciseSpan.UserKeyBounds()
@@ -2173,7 +2149,6 @@ func (d *DB) ingestApply(
 					}
 					newFiles := applyExciseToVersionEdit(ve, m, leftTable, rightTable, layer.Level())
 					replacedTables[m.TableNum] = newFiles
-					updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
 				}
 			}
 			if d.FormatMajorVersion() >= FormatExciseBoundsRecord {
@@ -2186,7 +2161,7 @@ func (d *DB) ingestApply(
 		if len(filesToSplit) > 0 {
 			// For the same reasons as the above call to excise, we hold the db mutex
 			// while calling this method.
-			if err := d.ingestSplit(ctx, ve, updateLevelMetricsOnExcise, filesToSplit, replacedTables); err != nil {
+			if err := d.ingestSplit(ctx, ve, filesToSplit, replacedTables); err != nil {
 				return versionUpdate{}, err
 			}
 		}

version_set_test.go

@@ -147,23 +147,10 @@ func TestVersionSet(t *testing.T) {
 				createFile(ve.CreatedBackingTables[i].DiskFileNum)
 			}
 
-			fileMetrics := newFileMetrics(ve.NewTables)
-			for de, f := range ve.DeletedTables {
-				lm := fileMetrics[de.Level]
-				if lm == nil {
-					lm = &LevelMetrics{}
-					fileMetrics[de.Level] = lm
-				}
-				lm.TablesCount--
-				lm.TablesSize -= int64(f.Size)
-				lm.EstimatedReferencesSize -= f.EstimatedReferenceSize()
-			}
-
 			mu.Lock()
 			_, err = vs.UpdateVersionLocked(func() (versionUpdate, error) {
 				return versionUpdate{
 					VE:                      ve,
-					Metrics:                 fileMetrics,
 					ForceManifestRotation:   rand.IntN(3) == 0,
 					InProgressCompactionsFn: func() []compactionInfo { return nil },
 				}, nil