db: add blob file rewrite compaction

Introduce a new compaction kind: blob-file rewrite compactions. These
compactions pick a physical blob file with unreferenced values, rewrite it and
output a new physical blob file. This commit adds the compaction and compaction
picking mechanics without the sophisticated blob file rewrite. For now these
rewrite compactions only copy the blob file verbatim.

Informs #112.

Author: jbowens

blob_rewrite.go

@@ -194,14 +194,15 @@ func TestBlobRewrite(t *testing.T) {
 
 				blobWriter := blob.NewFileWriter(fn, outputWritable, blob.FileWriterOptions{})
 				rewriter := newBlobFileRewriter(mockFC, block.ReadEnv{}, blobWriter, sstables, inputBlob)
-				err = rewriter.Rewrite()
+				stats, err := rewriter.Rewrite(context.Background())
 				if err != nil {
 					fmt.Fprintf(&buf, "rewrite error: %v\n", err)
 				} else {
 					fmt.Fprintf(&buf, "Successfully rewrote blob file %s to %s\n",
 						targetBlobFileStr, fn.String())
 					fmt.Fprintf(&buf, "Input SSTables: %v\n", sstableFileNums)
 					fmt.Fprintf(&buf, "SSTables with blob references: %d\n", len(sstables))
+					fmt.Fprintln(&buf, stats)
 				}
 
 				return buf.String()

blob_rewrite_test.go

@@ -151,6 +151,7 @@ const (
 	compactionKindTombstoneDensity
 	compactionKindRewrite
 	compactionKindIngestedFlushable
+	compactionKindBlobFileRewrite
 )
 
 func (k compactionKind) String() string {
@@ -175,6 +176,8 @@ func (k compactionKind) String() string {
 		return "ingested-flushable"
 	case compactionKindCopy:
 		return "copy"
+	case compactionKindBlobFileRewrite:
+		return "blob-file-rewrite"
 	}
 	return "?"
 }
@@ -410,8 +413,9 @@ func (c *tableCompaction) IsFlush() bool                      { return len(c.flu
 func (c *tableCompaction) Info() compactionInfo {
 	info := compactionInfo{
 		versionEditApplied: c.versionEditApplied,
+		kind:               c.kind,
 		inputs:             c.inputs,
-		bounds:             c.bounds,
+		bounds:             &c.bounds,
 		outputLevel:        -1,
 	}
 	if c.outputLevel != nil {
@@ -1774,7 +1778,8 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 				// doing a [c,d) excise at the same time as this compaction, we will have
 				// to error out the whole compaction as we can't guarantee it hasn't/won't
 				// write a file overlapping with the excise span.
-				if c2.Bounds().Overlaps(d.cmp, &exciseBounds) {
+				bounds := c2.Bounds()
+				if bounds != nil && bounds.Overlaps(d.cmp, &exciseBounds) {
 					c2.Cancel()
 				}
 			}
@@ -3623,6 +3628,8 @@ func getDiskWriteCategoryForCompaction(opts *Options, kind compactionKind) vfs.D
 		return "sql-row-spill"
 	} else if kind == compactionKindFlush {
 		return "pebble-memtable-flush"
+	} else if kind == compactionKindBlobFileRewrite {
+		return "pebble-blob-file-rewrite"
 	} else {
 		return "pebble-compaction"
 	}

compaction.go

@@ -93,9 +93,13 @@ type compactionInfo struct {
 	// been committed. The compaction may still be in-progress deleting newly
 	// obsolete files.
 	versionEditApplied bool
-	inputs             []compactionLevel
-	outputLevel        int
-	bounds             base.UserKeyBounds
+	// kind indicates the kind of compaction.
+	kind        compactionKind
+	inputs      []compactionLevel
+	outputLevel int
+	// bounds may be nil if the compaction does not involve sstables
+	// (specifically, a blob file rewrite).
+	bounds *base.UserKeyBounds
 }
 
 func (info compactionInfo) String() string {
@@ -1448,6 +1452,12 @@ func (p *compactionPickerByScore) pickAutoNonScore(env compactionEnv) (pc picked
 		return pc
 	}
 
+	// Check for blob file rewrites. These are low-priority compactions because
+	// they don't help us keep up with writes, just reclaim disk space.
+	if pc := p.pickBlobFileRewriteCompaction(env); pc != nil {
+		return pc
+	}
+
 	if pc := p.pickReadTriggeredCompaction(env); pc != nil {
 		return pc
 	}
@@ -1662,6 +1672,51 @@ func (p *compactionPickerByScore) pickRewriteCompaction(
 	return nil
 }
 
+// pickBlobFileRewriteCompaction looks for compactions of blob files that
+// can be rewritten to reclaim disk space.
+func (p *compactionPickerByScore) pickBlobFileRewriteCompaction(
+	env compactionEnv,
+) (pc *pickedBlobFileCompaction) {
+	aggregateStats, heuristicStats := p.latestVersionState.blobFiles.Stats()
+	if heuristicStats.CountFilesEligible == 0 && heuristicStats.CountFilesTooRecent == 0 {
+		// No blob files with any garbage to rewrite.
+		return nil
+	}
+	policy := p.opts.Experimental.ValueSeparationPolicy()
+	if policy.TargetGarbageRatio >= 1.0 {
+		// Blob file rewrite compactions are disabled.
+		return nil
+	}
+	garbagePct := float64(aggregateStats.ValueSize-aggregateStats.ReferencedValueSize) /
+		float64(aggregateStats.ValueSize)
+	if garbagePct <= policy.TargetGarbageRatio {
+		// Not enough garbage to warrant a rewrite compaction.
+		return nil
+	}
+
+	// Check if there is an ongoing blob file rewrite compaction. If there is,
+	// don't schedule a new one.
+	for _, c := range env.inProgressCompactions {
+		if c.kind == compactionKindBlobFileRewrite {
+			return nil
+		}
+	}
+
+	candidate, ok := p.latestVersionState.blobFiles.ReplacementCandidate()
+	if !ok {
+		// None meet the heuristic.
+		return nil
+	}
+	// Add a reference to the version. The compaction will release the reference
+	// when it completes.
+	p.vers.Ref()
+	return &pickedBlobFileCompaction{
+		vers:              p.vers,
+		file:              candidate,
+		referencingTables: p.latestVersionState.blobFiles.ReferencingTables(candidate.FileID),
+	}
+}
+
 // pickTombstoneDensityCompaction looks for a compaction that eliminates
 // regions of extremely high point tombstone density. For each level, it picks
 // a file where the ratio of tombstone-dense blocks is at least

compaction_picker.go

@@ -267,11 +267,13 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 						break
 					}
 					fmt.Fprintf(&b, "L%d->L%d: %.1f\n", pc.startLevel.level, pc.outputLevel.level, pc.score)
+					bounds := pc.bounds.Clone()
 					inProgress = append(inProgress, compactionInfo{
 						inputs:      pc.inputs,
 						outputLevel: pc.outputLevel.level,
-						bounds:      pc.bounds.Clone(),
+						bounds:      &bounds,
 					})
+
 					if pc.outputLevel.level == 0 {
 						// Once we pick one L0->L0 compaction, we'll keep on doing so
 						// because the test isn't marking files as Compacting.
@@ -520,7 +522,13 @@ func TestCompactionPickerL0(t *testing.T) {
 							return fmt.Sprintf("cannot find compaction file %s", tableNum)
 						}
 						compactFile.CompactionState = manifest.CompactionStateCompacting
-						info.bounds = info.bounds.Union(DefaultComparer.Compare, compactFile.UserKeyBounds())
+						var bounds base.UserKeyBounds
+						if info.bounds != nil {
+							bounds = info.bounds.Union(DefaultComparer.Compare, compactFile.UserKeyBounds())
+						} else {
+							bounds = compactFile.UserKeyBounds()
+						}
+						info.bounds = &bounds
 						compactionFiles[level] = append(compactionFiles[level], compactFile)
 					}
 				}
@@ -745,7 +753,13 @@ func TestCompactionPickerConcurrency(t *testing.T) {
 							return fmt.Sprintf("cannot find compaction file %s", tableNum)
 						}
 						compactFile.CompactionState = manifest.CompactionStateCompacting
-						info.bounds = info.bounds.Union(DefaultComparer.Compare, compactFile.UserKeyBounds())
+						var bounds base.UserKeyBounds
+						if info.bounds != nil {
+							bounds = info.bounds.Union(DefaultComparer.Compare, compactFile.UserKeyBounds())
+						} else {
+							bounds = compactFile.UserKeyBounds()
+						}
+						info.bounds = &bounds
 						compactionFiles[level] = append(compactionFiles[level], compactFile)
 					}
 				}

compaction_picker_test.go

@@ -142,10 +142,12 @@ func init() {
 		compactionOptionalAndPriority{optional: true, priority: 60}
 	scheduledCompactionMap[compactionKindElisionOnly] =
 		compactionOptionalAndPriority{optional: true, priority: 50}
-	scheduledCompactionMap[compactionKindRead] =
+	scheduledCompactionMap[compactionKindBlobFileRewrite] =
 		compactionOptionalAndPriority{optional: true, priority: 40}
-	scheduledCompactionMap[compactionKindRewrite] =
+	scheduledCompactionMap[compactionKindRead] =
 		compactionOptionalAndPriority{optional: true, priority: 30}
+	scheduledCompactionMap[compactionKindRewrite] =
+		compactionOptionalAndPriority{optional: true, priority: 20}
 }
 
 // noopGrantHandle is used in cases that don't interact with a CompactionScheduler.

compaction_scheduler.go

@@ -1781,8 +1781,8 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 				opts.TargetFileSizes[i] = opts.TargetFileSizes[i-1] * 2
 			}
 		case "value-separation":
-			if len(cmdArg.Vals) != 3 {
-				return errors.New("value-separation-policy expects 3 arguments: (enabled, minimum-size, max-blob-reference-depth)")
+			if len(cmdArg.Vals) != 5 {
+				return errors.New("value-separation expects 5 arguments: (enabled, minimum-size, max-blob-reference-depth, rewrite-minimum-age, target-garbage-ratio)")
 			}
 			var policy ValueSeparationPolicy
 			var err error
@@ -1798,6 +1798,14 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 			if err != nil {
 				return err
 			}
+			policy.RewriteMinimumAge, err = time.ParseDuration(cmdArg.Vals[3])
+			if err != nil {
+				return err
+			}
+			policy.TargetGarbageRatio, err = strconv.ParseFloat(cmdArg.Vals[4], 64)
+			if err != nil {
+				return err
+			}
 			opts.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy {
 				return policy
 			}

data_test.go

@@ -2860,7 +2860,7 @@ func (d *DB) getEarliestUnflushedSeqNumLocked() base.SeqNum {
 	return seqNum
 }
 
-func (d *DB) getInProgressCompactionInfoLocked(finishing *tableCompaction) (rv []compactionInfo) {
+func (d *DB) getInProgressCompactionInfoLocked(finishing compaction) (rv []compactionInfo) {
 	for c := range d.mu.compact.inProgress {
 		if !c.IsFlush() && (finishing == nil || c != finishing) {
 			rv = append(rv, c.Info())
@@ -2888,7 +2888,7 @@ func inProgressL0Compactions(inProgress []compactionInfo) []manifest.L0Compactio
 			continue
 		}
 		compactions = append(compactions, manifest.L0Compaction{
-			Bounds:    info.bounds,
+			Bounds:    *info.bounds,
 			IsIntraL0: info.outputLevel == 0,
 		})
 	}

db.go

@@ -139,6 +139,66 @@ func (i BlobFileDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 	w.Printf("[JOB %d] blob file deleted %s", redact.Safe(i.JobID), i.FileNum)
 }
 
+// BlobFileRewriteInfo contains the info for a blob file rewrite event.
+type BlobFileRewriteInfo struct {
+	// JobID is the ID of the job.
+	JobID int
+	// Input contains the input tables for the compaction organized by level.
+	Input BlobFileInfo
+	// Output contains the output tables generated by the compaction. The output
+	// info is empty for the compaction begin event.
+	Output BlobFileInfo
+	// Duration is the time spent compacting, including reading and writing
+	// files.
+	Duration time.Duration
+	// TotalDuration is the total wall-time duration of the compaction,
+	// including applying the compaction to the database. TotalDuration is
+	// always ≥ Duration.
+	TotalDuration time.Duration
+	Done          bool
+	// Err is set only if Done is true. If non-nil, indicates that the compaction
+	// failed. Note that err can be ErrCancelledCompaction, which can happen
+	// during normal operation.
+	Err error
+}
+
+func (i BlobFileRewriteInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i BlobFileRewriteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	if i.Err != nil {
+		w.Printf("[JOB %d] blob file (%s, %s) rewrite error: %s",
+			redact.Safe(i.JobID), i.Input.BlobFileID, i.Input.DiskFileNum, i.Err)
+		return
+	}
+
+	if !i.Done {
+		w.Printf("[JOB %d] rewriting blob file %s (physical file %s)",
+			redact.Safe(i.JobID), i.Input.BlobFileID, i.Input.DiskFileNum)
+		return
+	}
+	w.Printf("[JOB %d] rewrote blob file (%s, %s) -> (%s, %s), in %.1fs (%.1fs total)",
+		redact.Safe(i.JobID), i.Input.BlobFileID, i.Input.DiskFileNum,
+		i.Output.BlobFileID, i.Output.DiskFileNum,
+		redact.Safe(i.Duration.Seconds()),
+		redact.Safe(i.TotalDuration.Seconds()))
+}
+
+// BlobFileInfo describes a blob file.
+type BlobFileInfo struct {
+	// BlobFileID is the logical ID of the blob file.
+	BlobFileID base.BlobFileID
+	// DiskFileNum is the file number of the blob file on disk.
+	DiskFileNum base.DiskFileNum
+	// Size is the physical size of the file in bytes.
+	Size uint64
+	// ValueSize is the pre-compressed size of the values in the blob file in
+	// bytes.
+	ValueSize uint64
+}
+
 // CompactionInfo contains the info for a compaction event.
 type CompactionInfo struct {
 	// JobID is the ID of the compaction job.
@@ -802,6 +862,12 @@ type EventListener struct {
 	// BlobFileDeleted is invoked after a blob file has been deleted.
 	BlobFileDeleted func(BlobFileDeleteInfo)
 
+	// BlobFileRewriteBegin is invoked when a blob file rewrite compaction begins.
+	BlobFileRewriteBegin func(BlobFileRewriteInfo)
+
+	// BlobFileRewriteEnd is invoked when a blob file rewrite compaction ends.
+	BlobFileRewriteEnd func(BlobFileRewriteInfo)
+
 	// DataCorruption is invoked when an on-disk corruption is detected. It should
 	// not block, as it is called synchronously in read paths.
 	DataCorruption func(DataCorruptionInfo)
@@ -907,6 +973,12 @@ func (l *EventListener) EnsureDefaults(logger Logger) {
 	if l.BlobFileDeleted == nil {
 		l.BlobFileDeleted = func(info BlobFileDeleteInfo) {}
 	}
+	if l.BlobFileRewriteBegin == nil {
+		l.BlobFileRewriteBegin = func(info BlobFileRewriteInfo) {}
+	}
+	if l.BlobFileRewriteEnd == nil {
+		l.BlobFileRewriteEnd = func(info BlobFileRewriteInfo) {}
+	}
 	if l.DataCorruption == nil {
 		if logger != nil {
 			l.DataCorruption = func(info DataCorruptionInfo) {
@@ -998,6 +1070,12 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		BlobFileDeleted: func(info BlobFileDeleteInfo) {
 			logger.Infof("%s", info)
 		},
+		BlobFileRewriteBegin: func(info BlobFileRewriteInfo) {
+			logger.Infof("%s", info)
+		},
+		BlobFileRewriteEnd: func(info BlobFileRewriteInfo) {
+			logger.Infof("%s", info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			logger.Errorf("%s", info)
 		},
@@ -1084,6 +1162,14 @@ func TeeEventListener(a, b EventListener) EventListener {
 			a.BlobFileDeleted(info)
 			b.BlobFileDeleted(info)
 		},
+		BlobFileRewriteBegin: func(info BlobFileRewriteInfo) {
+			a.BlobFileRewriteBegin(info)
+			b.BlobFileRewriteBegin(info)
+		},
+		BlobFileRewriteEnd: func(info BlobFileRewriteInfo) {
+			a.BlobFileRewriteEnd(info)
+			b.BlobFileRewriteEnd(info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			a.DataCorruption(info)
 			b.DataCorruption(info)

event.go

@@ -1034,7 +1034,8 @@ func ingestTargetLevel(
 			if tblCompaction.outputLevel == nil || level != tblCompaction.outputLevel.level {
 				continue
 			}
-			if metaBounds.Overlaps(cmp, tblCompaction.Bounds()) {
+			bounds := tblCompaction.Bounds()
+			if bounds != nil && metaBounds.Overlaps(cmp, bounds) {
 				overlaps = true
 				break
 			}
@@ -2115,7 +2116,8 @@ func (d *DB) ingestApply(
 				// doing a [c,d) excise at the same time as this compaction, we will have
 				// to error out the whole compaction as we can't guarantee it hasn't/won't
 				// write a file overlapping with the excise span.
-				if c.Bounds().Overlaps(d.cmp, &exciseBounds) {
+				bounds := c.Bounds()
+				if bounds != nil && bounds.Overlaps(d.cmp, &exciseBounds) {
 					c.Cancel()
 				}
 				// Check if this compaction's inputs have been replaced due to an