db: collect blob file properties

Collect blob file properties as part of initial table stats
collection. Properties for newly written blob files are populated
directly.

Author: RaduBerinde

blob_rewrite.go

@@ -308,6 +308,14 @@ func (d *DB) runBlobFileRewriteLocked(
 		return objstorage.ObjectMetadata{}, nil, err
 	}
 
+	physical := &manifest.PhysicalBlobFile{
+		FileNum:      objMeta.DiskFileNum,
+		Size:         stats.FileLen,
+		ValueSize:    stats.UncompressedValueBytes,
+		CreationTime: uint64(d.timeNow().Unix()),
+	}
+	physical.PopulateProperties(&stats.Properties)
+
 	ve := &manifest.VersionEdit{
 		DeletedBlobFiles: map[manifest.DeletedBlobFileEntry]*manifest.PhysicalBlobFile{
 			{
@@ -317,13 +325,8 @@ func (d *DB) runBlobFileRewriteLocked(
 		},
 		NewBlobFiles: []manifest.BlobFileMetadata{
 			{
-				FileID: c.input.FileID,
-				Physical: &manifest.PhysicalBlobFile{
-					FileNum:      objMeta.DiskFileNum,
-					Size:         stats.FileLen,
-					ValueSize:    stats.UncompressedValueBytes,
-					CreationTime: uint64(d.timeNow().Unix()),
-				},
+				FileID:   c.input.FileID,
+				Physical: physical,
 			},
 		},
 	}

data_test.go

@@ -903,6 +903,8 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 	}
 	d.mu.Lock()
 	defer d.mu.Unlock()
+	// Make sure the table stats finished loading.
+	d.waitTableStatsLocked()
 	for i := range snapshots {
 		s := &Snapshot{db: d}
 		s.seqNum = snapshots[i]
@@ -1206,7 +1208,11 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 		return nil, err
 	}
 	d.updateReadStateLocked(nil)
-	d.updateTableStatsLocked(ve.NewTables)
+
+	// Force a re-read of table and blob file stats.
+	d.mu.tableStats.loadedInitial = false
+	d.maybeCollectTableStatsLocked()
+	d.waitTableStatsLocked()
 
 	return d, nil
 }

format_major_version.go

@@ -117,7 +117,7 @@ const (
 	// FormatMinTableFormatPebblev1 is a format major version that guarantees that
 	// tables created by or ingested into the DB at or above this format major
 	// version will have a table format version of at least Pebblev1 (Block
-	// Properties).
+	// FileProperties).
 	// Deprecated.
 	_ // FormatMinTableFormatPebblev1
 

internal/genericcache/cache.go

@@ -107,7 +107,7 @@ type ValueRef[K Key, V any, InitOpts any] struct {
 }
 
 // Value returns the value. This method and the returned value can only be used
-// until ref.Close() is called.
+// until ref.Unref() is called.
 func (ref ValueRef[K, V, InitOpts]) Value() *V {
 	if invariants.Enabled && ref.value.err != nil {
 		panic("ValueRef with error")

internal/manifest/blob_metadata.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/strparse"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/redact"
 )
 
@@ -134,6 +135,10 @@ type PhysicalBlobFile struct {
 	// referencing TableMetadata is installed in a Version and decremented when
 	// that TableMetadata becomes obsolete.
 	refs atomic.Int32
+
+	propsValid atomic.Bool
+	// stats are populated exactly once.
+	props blob.FileProperties
 }
 
 // SafeFormat implements redact.SafeFormatter.
@@ -159,6 +164,27 @@ func (m *PhysicalBlobFile) UserKeyBounds() base.UserKeyBounds {
 	return base.UserKeyBounds{}
 }
 
+// Properties returns the blob file properties if they have been populated, or
+// nil and ok=false if they were not.
+//
+// The caller must not modify the returned stats.
+func (m *PhysicalBlobFile) Properties() (_ *blob.FileProperties, ok bool) {
+	if !m.propsValid.Load() {
+		return nil, false
+	}
+	return &m.props, true
+}
+
+// PopulateProperties populates the bob file properties. Can be called at most
+// once for a TableBacking.
+func (m *PhysicalBlobFile) PopulateProperties(props *blob.FileProperties) {
+	m.props = *props
+	oldPropsValid := m.propsValid.Swap(true)
+	if invariants.Enabled && oldPropsValid {
+		panic("props set twice")
+	}
+}
+
 // ref increments the reference count for the blob file.
 func (m *PhysicalBlobFile) ref() {
 	m.refs.Add(+1)

metrics.go

@@ -134,7 +134,7 @@ type LevelMetrics struct {
 
 	// Additional contains misc additional metrics that are not always printed.
 	Additional struct {
-		// The sum of Properties.ValueBlocksSize for all the sstables in this
+		// The sum of FileProperties.ValueBlocksSize for all the sstables in this
 		// level. Printed by LevelMetrics.format iff there is at least one level
 		// with a non-zero value.
 		ValueBlocksSize uint64

sstable/blob/blob.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/crc"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -76,8 +77,8 @@ const (
 	//  20      Checksum type (1 byte)
 	//  21      File format version (1 byte)
 	//  22:30   Original file number (8 bytes)
-	//  30:38   Properties block offset (8 bytes)
-	//  38:46   Properties block length (8 bytes)
+	//  30:38   FileProperties block offset (8 bytes)
+	//  38:46   FileProperties block length (8 bytes)
 	//  46:54   Metaindex block offset (8 bytes); optional (for future use)
 	//  54:62   Metaindex block length (8 bytes); optional (for future use)
 	//  62:70   Blob file magic string (8 bytes)
@@ -123,6 +124,7 @@ type FileWriterStats struct {
 	ValueCount             uint32
 	UncompressedValueBytes uint64
 	FileLen                uint64
+	Properties             FileProperties
 }
 
 // String implements the fmt.Stringer interface.
@@ -325,14 +327,14 @@ func (w *FileWriter) Close() (FileWriterStats, error) {
 		}
 
 		// Write the properties and the v2 footer if the file format is v2.
+		w.stats.Properties = FileProperties{
+			CompressionStats: w.physBlockMaker.Compressor.Stats().Clone(),
+		}
 		var propBlockHandle block.Handle
 		if w.format >= FileFormatV2 {
 			var cw colblk.KeyValueBlockWriter
 			cw.Init()
-			p := Properties{
-				CompressionStats: w.physBlockMaker.Compressor.Stats().String(),
-			}
-			p.writeTo(&cw)
+			w.stats.Properties.writeTo(&cw)
 			propBlockHandle, err = w.writeMetadataBlock(cw.Finish(cw.Rows()))
 			if err != nil {
 				return err
@@ -616,40 +618,45 @@ func (r *FileReader) Layout() (string, error) {
 	return buf.String(), nil
 }
 
-type Properties struct {
-	CompressionStats string
+type FileProperties struct {
+	CompressionStats block.CompressionStats
 }
 
 // String returns any set properties, one per line.
-func (p *Properties) String() string {
+func (p *FileProperties) String() string {
 	var buf bytes.Buffer
-	if p.CompressionStats != "" {
-		fmt.Fprintf(&buf, "%s: %s\n", propertyKeyCompressionStats, p.CompressionStats)
+	if !p.CompressionStats.IsEmpty() {
+		fmt.Fprintf(&buf, "%s: %s\n", propertyKeyCompressionStats, p.CompressionStats.String())
 	}
 	return buf.String()
 }
 
-func (p *Properties) set(key []byte, value []byte) {
+func (p *FileProperties) set(key []byte, value []byte) {
 	switch string(key) {
 	case propertyKeyCompressionStats:
-		p.CompressionStats = string(value)
+		var err error
+		p.CompressionStats, err = block.ParseCompressionStats(string(value))
+		if invariants.Enabled && err != nil {
+			panic(errors.AssertionFailedf("pebble: error parsing blob file compression stats %q", string(value)))
+		}
+
 	default:
 		// Ignore unknown properties (for forward compatibility).
 	}
 }
 
-func (p *Properties) writeTo(w *colblk.KeyValueBlockWriter) {
-	if p.CompressionStats != "" {
-		w.AddKV([]byte(propertyKeyCompressionStats), []byte(p.CompressionStats))
+func (p *FileProperties) writeTo(w *colblk.KeyValueBlockWriter) {
+	if !p.CompressionStats.IsEmpty() {
+		w.AddKV([]byte(propertyKeyCompressionStats), []byte(p.CompressionStats.String()))
 	}
 }
 
 const propertyKeyCompressionStats = "compression_stats"
 
 // ReadProperties reads the properties block from the file, if it exists.
-func (r *FileReader) ReadProperties(ctx context.Context) (Properties, error) {
+func (r *FileReader) ReadProperties(ctx context.Context) (FileProperties, error) {
 	if r.footer.format < FileFormatV2 {
-		return Properties{}, nil
+		return FileProperties{}, nil
 	}
 	// We don't want the property block to go into the block cache, so we use a
 	// buffer pool.
@@ -661,12 +668,12 @@ func (r *FileReader) ReadProperties(ctx context.Context) (Properties, error) {
 		func(*block.Metadata, []byte) error { return nil },
 	)
 	if err != nil {
-		return Properties{}, err
+		return FileProperties{}, err
 	}
 	defer b.Release()
 	var decoder colblk.KeyValueBlockDecoder
 	decoder.Init(b.BlockData())
-	var p Properties
+	var p FileProperties
 	for k, v := range decoder.All() {
 		p.set(k, v)
 	}

sstable/block/compression_stats.go

@@ -154,6 +154,20 @@ func (c CompressionStats) String() string {
 	return buf.String()
 }
 
+// Clone returns a copy of the CompressionStats.
+func (c *CompressionStats) Clone() CompressionStats {
+	var out CompressionStats
+	out.noCompressionBytes = c.noCompressionBytes
+	out.fastest = c.fastest
+	if len(c.others) > 0 {
+		out.others = make(map[compression.Setting]CompressionStatsForSetting, len(c.others))
+		for s, cs := range c.others {
+			out.others[s] = cs
+		}
+	}
+	return out
+}
+
 // Scale the stats by (size/backingSize). Used to obtain an approximation of the
 // stats for a virtual table.
 //

table_stats.go

@@ -99,7 +99,6 @@ func (d *DB) collectTableStats() bool {
 	d.mu.tableStats.pending = nil
 	d.mu.tableStats.loading = true
 	jobID := d.newJobIDLocked()
-	loadedInitial := d.mu.tableStats.loadedInitial
 	// Drop DB.mu before performing IO.
 	d.mu.Unlock()
 
@@ -111,22 +110,29 @@ func (d *DB) collectTableStats() bool {
 	rs := d.loadReadState()
 	var collected []collectedStats
 	var hints []deleteCompactionHint
+	initialLoadCompleted := false
 	if len(pending) > 0 {
 		collected, hints = d.loadNewFileStats(ctx, rs, pending)
 	} else {
 		var moreRemain bool
 		var buf [maxTableStatsPerScan]collectedStats
-		collected, hints, moreRemain = d.scanReadStateTableStats(ctx, rs, buf[:0])
-		loadedInitial = !moreRemain
+		collected, hints, moreRemain = d.scanReadStateTableStats(ctx, rs.current, buf[:0])
+		if !moreRemain {
+			// Once we're done with table stats, load blob file properties.
+			moreRemain = d.scanBlobFileProperties(ctx, rs.current, maxTableStatsPerScan-len(collected))
+			if !moreRemain {
+				initialLoadCompleted = true
+			}
+		}
 	}
 	rs.unref()
 
 	// Update the TableMetadata with the loaded stats while holding d.mu.
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	d.mu.tableStats.loading = false
-	if loadedInitial && !d.mu.tableStats.loadedInitial {
-		d.mu.tableStats.loadedInitial = loadedInitial
+	if initialLoadCompleted && !d.mu.tableStats.loadedInitial {
+		d.mu.tableStats.loadedInitial = true
 		d.opts.EventListener.TableStatsLoaded(TableStatsInfo{
 			JobID: int(jobID),
 		})
@@ -216,12 +222,12 @@ func (d *DB) loadNewFileStats(
 // are no pending new files, but there might be files that existed at Open for
 // which we haven't loaded table stats.
 func (d *DB) scanReadStateTableStats(
-	ctx context.Context, rs *readState, fill []collectedStats,
-) ([]collectedStats, []deleteCompactionHint, bool) {
-	moreRemain := false
+	ctx context.Context, version *manifest.Version, fill []collectedStats,
+) (_ []collectedStats, _ []deleteCompactionHint, moreRemain bool) {
 	var hints []deleteCompactionHint
 	sizesChecked := make(map[base.DiskFileNum]struct{})
-	for l, levelMetadata := range rs.current.Levels {
+	// TODO(radu): an O(#tables) scan every time could be problematic.
+	for l, levelMetadata := range version.Levels {
 		for f := range levelMetadata.All() {
 			// NB: Only the active stats collection job updates f.Stats for active
 			// files, and we ensure only one goroutine runs it at a time through
@@ -278,7 +284,7 @@ func (d *DB) scanReadStateTableStats(
 				sizesChecked[f.TableBacking.DiskFileNum] = struct{}{}
 			}
 
-			stats, newHints, err := d.loadTableStats(ctx, rs.current, l, f)
+			stats, newHints, err := d.loadTableStats(ctx, version, l, f)
 			if err != nil {
 				// Set `moreRemain` so we'll try again.
 				moreRemain = true
@@ -295,6 +301,43 @@ func (d *DB) scanReadStateTableStats(
 	return fill, hints, moreRemain
 }
 
+// populateBlobFileProperties reads at most maxNum blob file properties for blob
+// files that don't have them populated. Returns false once all properties have
+// been populated.
+func (d *DB) scanBlobFileProperties(
+	ctx context.Context, version *manifest.Version, maxNum int,
+) (moreRemain bool) {
+	// TODO(radu): an O(#files) scan every time could be problematic.
+	// We could remember the last blob file ID and scan from there.
+	for f := range version.BlobFiles.All() {
+		if _, propsValid := f.Physical.Properties(); propsValid {
+			// Properties are already populated.
+			continue
+		}
+		if maxNum == 0 {
+			return moreRemain
+		}
+		v, err := d.fileCache.findOrCreateBlob(ctx, f.Physical, block.InitFileReadStats{})
+		if err != nil {
+			moreRemain = true
+			continue
+		}
+		blobReader := v.Value().mustBlob()
+		blobProps, err := blobReader.ReadProperties(ctx)
+		v.Unref()
+		if err != nil {
+			moreRemain = true
+			continue
+		}
+		// It is ok to call PopulateProperties here because this function runs as
+		// part of a table statistics job, and at most one goroutine runs this at a
+		// time (see d.mu.tableStats.loading).
+		f.Physical.PopulateProperties(&blobProps)
+		maxNum--
+	}
+	return moreRemain
+}
+
 func (d *DB) loadTableStats(
 	ctx context.Context, v *manifest.Version, level int, meta *manifest.TableMetadata,
 ) (manifest.TableStats, []deleteCompactionHint, error) {
@@ -525,6 +568,9 @@ func (d *DB) estimateSizesBeneath(
 					if err != nil {
 						return err
 					}
+					// It is ok to call PopulateProperties here because this function runs as part of
+					// a table statistics job, and at most one goroutine runs this at a
+					// time (see d.mu.tableStats.loading).
 					backingProps = tableBeneath.TableBacking.PopulateProperties(&loadedProps)
 					return nil
 				})

testdata/compaction/value_separation

@@ -158,7 +158,7 @@ ITERATORS
         block cache        |         file cache         |    filter    |  sst iters  |  snapshots
      entries |    hit rate |      entries |    hit rate |         util |        open |        open
 -------------+-------------+--------------+-------------+--------------+-------------+------------
-   5 (1.6KB) |       81.8% |     1 (392B) |       89.2% |         0.0% |           0 |           0
+   5 (1.6KB) |       81.8% |     1 (392B) |       89.6% |         0.0% |           0 |           0
 
 FILES                 tables                       |       blob files        |     blob values
    stats prog |    backing |                zombie |       live |     zombie |  total |      refed