db: don't store table fromat in compaction

We currently store a table format in compaction. In practice, this is
always `d.TableFormat()`, and we don't even end up using it when we
write the output (we use `d.TableFormat()` again).

This change removes the table format from the compaction to simplify
things a bit. This allows adding a simpler `DB.makeWriterOptions()`
(which makes future extensions easier).

Author: RaduBerinde

blob_rewrite.go

@@ -320,7 +320,7 @@ func (d *DB) runBlobFileRewriteLocked(
 	if err != nil {
 		return objstorage.ObjectMetadata{}, nil, err
 	}
-	// Initialize a blob file rewriter. We pass L6 to MakeBlobWriterOptions.
+	// Initialize a blob file rewriter. We pass L6 to makeBlobWriterOptions.
 	// There's no single associated level with a blob file. A long-lived blob
 	// file that gets rewritten is likely to mostly be referenced from L6.
 	// TODO(jackson): Consider refactoring to remove the level association.
@@ -329,7 +329,7 @@ func (d *DB) runBlobFileRewriteLocked(
 		env,
 		objMeta.DiskFileNum,
 		writable,
-		d.opts.MakeBlobWriterOptions(6, d.BlobFileFormat()),
+		d.makeBlobWriterOptions(6),
 		c.referencingTables,
 		c.input,
 	)

compaction.go

@@ -30,7 +30,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
-	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -251,7 +250,7 @@ type tableCompaction struct {
 	// b) rewrite blob files: The compaction will write eligible values to new
 	// blob files. This consumes more write bandwidth because all values are
 	// rewritten. However it restores locality.
-	getValueSeparation func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation
+	getValueSeparation func(JobID, *tableCompaction) compact.ValueSeparation
 
 	// startLevel is the level that is being compacted. Inputs from startLevel
 	// and outputLevel will be merged to produce a set of outputLevel files.
@@ -336,7 +335,6 @@ type tableCompaction struct {
 
 	grantHandle CompactionGrantHandle
 
-	tableFormat   sstable.TableFormat
 	objCreateOpts objstorage.CreateOptions
 
 	annotations []string
@@ -544,7 +542,7 @@ func (c *tableCompaction) makeInfo(jobID JobID) CompactionInfo {
 	return info
 }
 
-type getValueSeparation func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation
+type getValueSeparation func(JobID, *tableCompaction) compact.ValueSeparation
 
 // newCompaction constructs a compaction from the provided picked compaction.
 //
@@ -556,7 +554,7 @@ func newCompaction(
 	beganAt time.Time,
 	provider objstorage.Provider,
 	grantHandle CompactionGrantHandle,
-	tableFormat sstable.TableFormat,
+	preferSharedStorage bool,
 	getValueSeparation getValueSeparation,
 ) *tableCompaction {
 	c := &tableCompaction{
@@ -574,7 +572,6 @@ func newCompaction(
 			picker:  pc.pickerMetrics,
 		},
 		grantHandle: grantHandle,
-		tableFormat: tableFormat,
 	}
 	// Acquire a reference to the version to ensure that files and in-memory
 	// version state necessary for reading files remain available. Ignoring
@@ -615,8 +612,6 @@ func newCompaction(
 	)
 	c.kind = pc.kind
 
-	preferSharedStorage := tableFormat >= FormatMinForSharedObjects.MaxTableFormat() &&
-		remote.ShouldCreateShared(opts.Experimental.CreateOnShared, c.outputLevel.level)
 	c.maybeSwitchToMoveOrCopy(preferSharedStorage, provider)
 	c.objCreateOpts = objstorage.CreateOptions{
 		PreferSharedStorage: preferSharedStorage,
@@ -833,7 +828,7 @@ func newFlush(
 	baseLevel int,
 	flushing flushableList,
 	beganAt time.Time,
-	tableFormat sstable.TableFormat,
+	preferSharedStorage bool,
 	getValueSeparation getValueSeparation,
 ) (*tableCompaction, error) {
 	c := &tableCompaction{
@@ -845,7 +840,6 @@ func newFlush(
 		maxOutputFileSize:  math.MaxUint64,
 		maxOverlapBytes:    math.MaxUint64,
 		grantHandle:        noopGrantHandle{},
-		tableFormat:        tableFormat,
 		metrics: compactionMetrics{
 			beganAt: beganAt,
 		},
@@ -872,8 +866,6 @@ func newFlush(
 		}
 	}
 
-	preferSharedStorage := tableFormat >= FormatMinForSharedObjects.MaxTableFormat() &&
-		remote.ShouldCreateShared(opts.Experimental.CreateOnShared, c.outputLevel.level)
 	c.objCreateOpts = objstorage.CreateOptions{
 		PreferSharedStorage: preferSharedStorage,
 		WriteCategory:       getDiskWriteCategoryForCompaction(opts, c.kind),
@@ -1722,8 +1714,16 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		}
 	}
 
-	c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.latest.l0Organizer,
-		d.mu.versions.picker.getBaseLevel(), d.mu.mem.queue[:n], d.timeNow(), d.TableFormat(), d.determineCompactionValueSeparation)
+	c, err := newFlush(
+		d.opts,
+		d.mu.versions.currentVersion(),
+		d.mu.versions.latest.l0Organizer,
+		d.mu.versions.picker.getBaseLevel(),
+		d.mu.mem.queue[:n],
+		d.timeNow(),
+		d.shouldCreateShared(0),
+		d.determineCompactionValueSeparation,
+	)
 	if err != nil {
 		return 0, err
 	}
@@ -2921,12 +2921,7 @@ func (d *DB) runCopyCompaction(
 			}
 		}()
 
-		w, _, err := d.objProvider.Create(
-			ctx, base.FileTypeTable, newMeta.TableBacking.DiskFileNum,
-			objstorage.CreateOptions{
-				PreferSharedStorage: d.shouldCreateShared(c.outputLevel.level),
-			},
-		)
+		w, _, err := d.objProvider.Create(ctx, base.FileTypeTable, newMeta.TableBacking.DiskFileNum, c.objCreateOpts)
 		if err != nil {
 			return nil, compact.Stats{}, []compact.OutputBlob{}, err
 		}
@@ -2954,7 +2949,7 @@ func (d *DB) runCopyCompaction(
 			// or update category stats).
 			wrote, err = sstable.CopySpan(ctx,
 				src, r, c.startLevel.level,
-				w, d.opts.MakeWriterOptions(c.outputLevel.level, d.TableFormat()),
+				w, d.makeWriterOptions(c.outputLevel.level),
 				start, end,
 			)
 			return err
@@ -3306,9 +3301,9 @@ func (d *DB) runDefaultTableCompaction(
 	defer d.mu.Lock()
 
 	// Determine whether we should separate values into blob files.
-	valueSeparation := c.getValueSeparation(jobID, c, c.tableFormat)
+	valueSeparation := c.getValueSeparation(jobID, c)
 
-	result := d.compactAndWrite(jobID, c, snapshots, c.tableFormat, valueSeparation)
+	result := d.compactAndWrite(jobID, c, snapshots, valueSeparation)
 	if result.Err == nil {
 		ve, result.Err = c.makeVersionEdit(result)
 	}
@@ -3352,7 +3347,6 @@ func (d *DB) compactAndWrite(
 	jobID JobID,
 	c *tableCompaction,
 	snapshots compact.Snapshots,
-	tableFormat sstable.TableFormat,
 	valueSeparation compact.ValueSeparation,
 ) (result compact.Result) {
 	suggestedCacheReaders := blob.SuggestedCachedReaders(len(c.inputs))
@@ -3469,7 +3463,7 @@ func (d *DB) compactAndWrite(
 			spanPolicyValid = true
 		}
 
-		writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, tableFormat)
+		writerOpts := d.makeWriterOptions(c.outputLevel.level)
 		if spanPolicy.DisableValueSeparationBySuffix {
 			writerOpts.DisableValueBlocks = true
 		}

compaction_picker.go

@@ -248,7 +248,7 @@ func (pc *pickedTableCompaction) ConstructCompaction(
 		d.timeNow(),
 		d.ObjProvider(),
 		grantHandle,
-		d.TableFormat(),
+		d.shouldCreateShared(pc.outputLevel.level),
 		d.determineCompactionValueSeparation)
 }
 

compaction_picker_test.go

@@ -671,7 +671,7 @@ func TestCompactionPickerL0(t *testing.T) {
 			var result strings.Builder
 			if ptc != nil {
 				checkClone(t, ptc)
-				c := newCompaction(ptc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+				c := newCompaction(ptc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 				fmt.Fprintf(&result, "L%d -> L%d\n", ptc.startLevel.level, ptc.outputLevel.level)
 				fmt.Fprintf(&result, "L%d: %s\n", ptc.startLevel.level, tableNums(ptc.startLevel.files))
 				if !ptc.outputLevel.files.Empty() {
@@ -713,6 +713,8 @@ func TestCompactionPickerL0(t *testing.T) {
 	})
 }
 
+const noSharedStorage = false
+
 func TestCompactionPickerConcurrency(t *testing.T) {
 	opts := DefaultOptions()
 	opts.Experimental.L0CompactionConcurrency = 1
@@ -803,7 +805,7 @@ func TestCompactionPickerConcurrency(t *testing.T) {
 			var result strings.Builder
 			fmt.Fprintf(&result, "picker.getCompactionConcurrency: %d\n", allowedCompactions)
 			if pc != nil {
-				c := newCompaction(ptc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+				c := newCompaction(ptc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 				fmt.Fprintf(&result, "L%d -> L%d\n", ptc.startLevel.level, ptc.outputLevel.level)
 				fmt.Fprintf(&result, "L%d: %s\n", ptc.startLevel.level, tableNums(ptc.startLevel.files))
 				if !ptc.outputLevel.files.Empty() {

compaction_test.go

@@ -545,7 +545,7 @@ func TestPickCompaction(t *testing.T) {
 		pc, got := vs.picker.pickAutoScore(compactionEnv{diskAvailBytes: math.MaxUint64}), ""
 		if pc != nil {
 			tableCompaction := pc.(*pickedTableCompaction)
-			c := newCompaction(tableCompaction, opts, time.Now(), nil /* provider */, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+			c := newCompaction(tableCompaction, opts, time.Now(), nil /* provider */, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 
 			gotStart := fileNums(c.startLevel.files)
 			gotML := ""
@@ -1609,7 +1609,7 @@ func TestCompactionOutputLevel(t *testing.T) {
 				d.ScanArgs(t, "start", &start)
 				d.ScanArgs(t, "base", &base)
 				pc := newPickedTableCompaction(opts, version, l0Organizer, start, defaultOutputLevel(start, base), base)
-				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 				return fmt.Sprintf("output=%d\nmax-output-file-size=%d\n",
 					c.outputLevel.level, c.maxOutputFileSize)
 
@@ -3551,7 +3551,7 @@ func TestTombstoneDensityCompactionMoveOptimization(t *testing.T) {
 
 	// Create the compaction.
 	tableCompaction := pc.(*pickedTableCompaction)
-	c := newCompaction(tableCompaction, opts, time.Now(), nil, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+	c := newCompaction(tableCompaction, opts, time.Now(), nil, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 
 	// The compaction should be converted to a move.
 	require.Equal(t, compactionKindMove, c.kind, "expected compaction to be optimized to a move")
@@ -3667,7 +3667,7 @@ func TestTombstoneDensityCompactionMoveOptimization_NoMoveWithOverlap(t *testing
 
 	// Create the compaction.
 	tableCompaction := pc.(*pickedTableCompaction)
-	c := newCompaction(tableCompaction, opts, time.Now(), nil, noopGrantHandle{}, sstable.TableFormatMinSupported, neverSeparateValues)
+	c := newCompaction(tableCompaction, opts, time.Now(), nil, noopGrantHandle{}, noSharedStorage, neverSeparateValues)
 
 	// The compaction should NOT be converted to a move.
 	require.NotEqual(t, compactionKindMove, c.kind, "move optimization should NOT apply when there is overlap in output level")

data_test.go

@@ -583,7 +583,7 @@ func runBuildRemoteCmd(td *datadriven.TestData, d *DB, storage remote.Storage) e
 	}
 	path := td.CmdArgs[0].String()
 
-	// Use TableFormatMax here and downgrade after, if necessary. This ensures
+	// Use d.TableFormat() here and downgrade after, if necessary. This ensures
 	// that all fields are set.
 	writeOpts := d.opts.MakeWriterOptions(0 /* level */, d.TableFormat())
 	if rand.IntN(4) == 0 {
@@ -648,7 +648,7 @@ func runBuildCmd(
 	}
 	path := td.CmdArgs[0].String()
 
-	writeOpts := d.opts.MakeWriterOptions(0 /* level */, d.TableFormat())
+	writeOpts := d.makeWriterOptions(0 /* level */)
 	if err := sstable.ParseWriterOptions(&writeOpts, td.CmdArgs[1:]...); err != nil {
 		return err
 	}
@@ -932,11 +932,11 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 			flushed:   make(chan struct{}),
 		}}
 		c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.latest.l0Organizer,
-			d.mu.versions.picker.getBaseLevel(), toFlush, time.Now(), d.TableFormat(), d.determineCompactionValueSeparation)
+			d.mu.versions.picker.getBaseLevel(), toFlush, time.Now(), d.shouldCreateShared(0), d.determineCompactionValueSeparation)
 		if err != nil {
 			return err
 		}
-		c.getValueSeparation = func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation {
+		c.getValueSeparation = func(JobID, *tableCompaction) compact.ValueSeparation {
 			return valueSeparator
 		}
 		// NB: define allows the test to exactly specify which keys go
@@ -1163,7 +1163,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 			writable, _, err := d.objProvider.Create(context.Background(), base.FileTypeBlob, fileNum, objstorage.CreateOptions{})
 			return writable, err
 		}
-		blobWriterOpts := d.opts.MakeBlobWriterOptions(0, d.BlobFileFormat())
+		blobWriterOpts := d.makeBlobWriterOptions(0)
 		fileStats, err := valueSeparator.bv.WriteFiles(newBlobObject, blobWriterOpts)
 		if err != nil {
 			return nil, err

download.go

@@ -446,7 +446,7 @@ func (d *DB) tryLaunchDownloadForFile(
 
 	download.numLaunchedDownloads++
 	doneCh = make(chan error, 1)
-	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{}, d.TableFormat(), d.determineCompactionValueSeparation)
+	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{}, d.shouldCreateShared(pc.outputLevel.level), d.determineCompactionValueSeparation)
 	c.isDownload = true
 	d.mu.compact.downloadingCount++
 	c.AddInProgressLocked(d)

options.go

@@ -2623,19 +2623,25 @@ func (o *Options) MakeWriterOptions(level int, format sstable.TableFormat) sstab
 	return writerOpts
 }
 
-// MakeBlobWriterOptions constructs blob.FileWriterOptions from the corresponding
-// options in the receiver.
-func (o *Options) MakeBlobWriterOptions(level int, format blob.FileFormat) blob.FileWriterOptions {
-	lo := o.Levels[level]
+// makeWriterOptions constructs sstable.WriterOptions for the specified level
+// using the current DB options and format.
+func (d *DB) makeWriterOptions(level int) sstable.WriterOptions {
+	return d.opts.MakeWriterOptions(level, d.TableFormat())
+}
+
+// makeBlobWriterOptions constructs blob.FileWriterOptions using the current DB
+// options and format.
+func (d *DB) makeBlobWriterOptions(level int) blob.FileWriterOptions {
+	lo := &d.opts.Levels[level]
 	return blob.FileWriterOptions{
-		Format:       format,
+		Format:       d.BlobFileFormat(),
 		Compression:  lo.Compression(),
 		ChecksumType: block.ChecksumTypeCRC32c,
 		FlushGovernor: block.MakeFlushGovernor(
 			lo.BlockSize,
 			lo.BlockSizeThreshold,
 			base.SizeClassAwareBlockSizeThreshold,
-			o.AllocatorSizeClasses,
+			d.opts.AllocatorSizeClasses,
 		),
 	}
 }

value_separation.go

@@ -24,17 +24,19 @@ import (
 // ValueStorageLatencyTolerant.
 const latencyTolerantMinimumSize = 10
 
-var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation {
+var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction) compact.ValueSeparation {
 	return compact.NeverSeparateValues{}
 }
 
 // determineCompactionValueSeparation determines whether a compaction should
 // separate values into blob files. It returns a compact.ValueSeparation
 // implementation that should be used for the compaction.
+//
+// It assumes that the compaction will write tables at d.TableFormat() or above.
 func (d *DB) determineCompactionValueSeparation(
-	jobID JobID, c *tableCompaction, tableFormat sstable.TableFormat,
+	jobID JobID, c *tableCompaction,
 ) compact.ValueSeparation {
-	if tableFormat < sstable.TableFormatPebblev7 || d.FormatMajorVersion() < FormatValueSeparation ||
+	if d.FormatMajorVersion() < FormatValueSeparation ||
 		d.opts.Experimental.ValueSeparationPolicy == nil {
 		return compact.NeverSeparateValues{}
 	}
@@ -61,7 +63,7 @@ func (d *DB) determineCompactionValueSeparation(
 				jobID, c.kind, c.outputLevel.level, &c.metrics.bytesWritten, c.objCreateOpts)
 		},
 		shortAttrExtractor: d.opts.Experimental.ShortAttributeExtractor,
-		writerOpts:         d.opts.MakeBlobWriterOptions(c.outputLevel.level, d.BlobFileFormat()),
+		writerOpts:         d.makeBlobWriterOptions(c.outputLevel.level),
 		minimumSize:        policy.MinimumSize,
 		globalMinimumSize:  policy.MinimumSize,
 		invalidValueCallback: func(userKey []byte, value []byte, err error) {