manifest: add L0Organizer

We introduce `L0Organizer` type which contains the logic for updating
`L0Sublevels`. A single `L0Organizer` per store keeps track of the
current L0 state (corresponding to the latest `Version`).

Author: RaduBerinde

compaction_test.go

@@ -41,9 +41,10 @@ import (
 )
 
 func newVersion(opts *Options, files [numLevels][]*tableMetadata) *version {
+	l0Organizer := manifest.NewL0Organizer(opts.Comparer, opts.FlushSplitBytes)
 	v := manifest.NewVersionForTesting(
 		opts.Comparer,
-		opts.FlushSplitBytes,
+		l0Organizer,
 		files)
 	if err := v.CheckOrdering(); err != nil {
 		panic(err)

download_test.go

@@ -33,7 +33,8 @@ func TestDownloadCursor(t *testing.T) {
 		case "define":
 			var err error
 			const flushSplitBytes = 10 * 1024 * 1024
-			vers, err = manifest.ParseVersionDebug(base.DefaultComparer, flushSplitBytes, td.Input)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, flushSplitBytes)
+			vers, err = manifest.ParseVersionDebug(base.DefaultComparer, l0Organizer, td.Input)
 			if err != nil {
 				td.Fatalf(t, "%v", err)
 			}
@@ -118,7 +119,8 @@ func TestDownloadTask(t *testing.T) {
 		case "define":
 			var err error
 			const flushSplitBytes = 10 * 1024 * 1024
-			vers, err = manifest.ParseVersionDebug(base.DefaultComparer, flushSplitBytes, td.Input)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, flushSplitBytes)
+			vers, err = manifest.ParseVersionDebug(base.DefaultComparer, l0Organizer, td.Input)
 			if err != nil {
 				td.Fatalf(t, "%v", err)
 			}

get_iter_test.go

@@ -429,7 +429,8 @@ func TestGetIter(t *testing.T) {
 
 			files[tt.level] = append(files[tt.level], meta)
 		}
-		v := manifest.NewVersionForTesting(cmp, 10<<20, files)
+		l0Organizer := manifest.NewL0Organizer(cmp, 10<<20 /*flushSplitBytes*/)
+		v := manifest.NewVersionForTesting(cmp, l0Organizer, files)
 		err := v.CheckOrdering()
 		if tc.badOrdering && err == nil {
 			t.Errorf("desc=%q: want bad ordering, got nil error", desc)

internal/compact/run_test.go

@@ -23,7 +23,8 @@ func TestTableSplitLimit(t *testing.T) {
 		case "define":
 			var flushSplitBytes int64
 			d.MaybeScanArgs(t, "flush-split-bytes", &flushSplitBytes)
-			v = testutils.CheckErr(manifest.ParseVersionDebug(base.DefaultComparer, flushSplitBytes, d.Input))
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, flushSplitBytes)
+			v = testutils.CheckErr(manifest.ParseVersionDebug(base.DefaultComparer, l0Organizer, d.Input))
 			buf.WriteString(v.String())
 			if v.Levels[0].Len() != 0 {
 				buf.WriteString("flush split keys:\n")

internal/compact/splitting_test.go

@@ -34,7 +34,8 @@ func TestOutputSplitter(t *testing.T) {
 					files[1] = append(files[1], f)
 				}
 			}
-			v := manifest.NewVersionForTesting(base.DefaultComparer, 64*1024, files)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, 64*1024 /* flushSplitBytes */)
+			v := manifest.NewVersionForTesting(base.DefaultComparer, l0Organizer, files)
 			if err := v.CheckOrdering(); err != nil {
 				d.Fatalf(t, "%v", err)
 			}

internal/compact/tombstone_elision_test.go

@@ -89,7 +89,8 @@ func TestSetupTombstoneElision(t *testing.T) {
 		switch td.Cmd {
 		case "define":
 			var err error
-			v, err = manifest.ParseVersionDebug(base.DefaultComparer, 64*1024, td.Input)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, 64*1024 /* flushSplitBytes */)
+			v, err = manifest.ParseVersionDebug(base.DefaultComparer, l0Organizer, td.Input)
 			if err != nil {
 				td.Fatalf(t, "%v", err)
 			}
@@ -129,7 +130,8 @@ func TestTombstoneElision(t *testing.T) {
 		switch td.Cmd {
 		case "define":
 			var err error
-			v, err = manifest.ParseVersionDebug(base.DefaultComparer, 64*1024, td.Input)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, 64*1024 /* flushSplitBytes */)
+			v, err = manifest.ParseVersionDebug(base.DefaultComparer, l0Organizer, td.Input)
 			if err != nil {
 				td.Fatalf(t, "%v", err)
 			}

internal/keyspan/keyspanimpl/level_iter_test.go

@@ -312,7 +312,9 @@ func TestLevelIterEquivalence(t *testing.T) {
 				amap[metas[i].FileNum] = metas[i]
 			}
 			b.AddedTables[6] = amap
-			v, err := b.Apply(manifest.NewVersion(base.DefaultComparer), 0, 0)
+			l0Organizer := manifest.NewL0Organizer(base.DefaultComparer, 0 /* flushSplitBytes */)
+			emptyVersion := manifest.NewInitialVersion(base.DefaultComparer, l0Organizer)
+			v, err := b.Apply(emptyVersion, l0Organizer, 0)
 			require.NoError(t, err)
 			levelIter.Init(
 				context.Background(),

internal/manifest/annotator_test.go

@@ -26,7 +26,8 @@ func makeTestVersion(numFiles int) (*Version, []*TableMetadata) {
 	var levelFiles [7][]*TableMetadata
 	levelFiles[6] = files
 
-	v := NewVersionForTesting(base.DefaultComparer, 0, levelFiles)
+	l0Organizer := NewL0Organizer(base.DefaultComparer, 0 /* flushSplitBytes */)
+	v := NewVersionForTesting(base.DefaultComparer, l0Organizer, levelFiles)
 	return v, files
 }
 

internal/manifest/l0_sublevels.go

@@ -249,7 +249,7 @@ type L0Sublevels struct {
 
 	fileBytes uint64
 	// All the L0 files, ordered from oldest to youngest.
-	levelMetadata *LevelMetadata
+	levelMetadata LevelMetadata
 
 	// The file intervals in increasing key order.
 	orderedIntervals []fileInterval
@@ -281,7 +281,7 @@ func NewL0Sublevels(
 	levelMetadata *LevelMetadata, cmp Compare, formatKey base.FormatKey, flushSplitMaxBytes int64,
 ) (*L0Sublevels, error) {
 	s := &L0Sublevels{cmp: cmp, formatKey: formatKey}
-	s.levelMetadata = levelMetadata
+	s.levelMetadata = *levelMetadata
 	keys := make([]intervalKeyTemp, 0, 2*s.levelMetadata.Len())
 	iter := levelMetadata.Iter()
 	for i, f := 0, iter.First(); f != nil; i, f = i+1, iter.Next() {
@@ -480,7 +480,7 @@ func (s *L0Sublevels) AddL0Files(
 	*newVal = *s
 
 	newVal.addL0FilesCalled = false
-	newVal.levelMetadata = levelMetadata
+	newVal.levelMetadata = *levelMetadata
 	// Deep copy levelFiles and Levels, as they are mutated and sorted below.
 	// Shallow copies of slices that we just append to, are okay.
 	newVal.levelFiles = make([][]*TableMetadata, len(s.levelFiles))
@@ -2073,3 +2073,164 @@ func (s *L0Sublevels) extendCandidateToRectangle(
 	}
 	return addedCount > 0
 }
+
+// L0Organizer keeps track of L0 state, including the subdivision into
+// sublevels.
+//
+// It is designed to be used as a singleton (per store) which gets updated as
+// the version changes. It is used to initialize L0-related Version fields.
+type L0Organizer struct {
+	cmp             base.Compare
+	formatKey       base.FormatKey
+	flushSplitBytes int64
+
+	// levelMetadata is the current L0.
+	levelMetadata LevelMetadata
+
+	// sublevels reflects the current L0.
+	sublevels *L0Sublevels
+}
+
+// NewL0Organizer creates the L0 organizer. The L0 organizer is responsible for
+// maintaining the current L0 state and is kept in-sync with the current Version.
+//
+// flushSplitBytes denotes the target number of bytes per sublevel in each flush
+// split interval (i.e. range between two flush split keys) in L0 sstables. When
+// set to zero, only a single sstable is generated by each flush. When set to a
+// non-zero value, flushes are split at points to meet L0's TargetFileSize, any
+// grandparent-related overlap options, and at boundary keys of L0 flush split
+// intervals (which are targeted to contain around FlushSplitBytes bytes in each
+// sublevel between pairs of boundary keys). Splitting sstables during flush
+// allows increased compaction flexibility and concurrency when those tables are
+// compacted to lower levels.
+func NewL0Organizer(comparer *base.Comparer, flushSplitBytes int64) *L0Organizer {
+	o := &L0Organizer{
+		cmp:             comparer.Compare,
+		formatKey:       comparer.FormatKey,
+		flushSplitBytes: flushSplitBytes,
+		levelMetadata:   MakeLevelMetadata(comparer.Compare, 0, nil),
+	}
+	var err error
+	o.sublevels, err = NewL0Sublevels(&o.levelMetadata, o.cmp, o.formatKey, o.flushSplitBytes)
+	if err != nil {
+		panic(errors.AssertionFailedf("error generating empty L0Sublevels: %s", err))
+	}
+	return o
+}
+
+// Sublevels returns the *L0Sublevels reflecting the current L0 state.
+func (o *L0Organizer) Sublevels() *L0Sublevels {
+	return o.sublevels
+}
+
+// Update the L0 organizer with the given L0 changes.
+func (o *L0Organizer) Update(
+	addedL0Tables map[base.FileNum]*TableMetadata,
+	deletedL0Tables map[base.FileNum]*TableMetadata,
+	newLevelMeta *LevelMetadata,
+) {
+	if invariants.Enabled && invariants.Sometimes(10) {
+		// Verify that newLevelMeta = m.levelMetadata + addedL0Tables - deletedL0Tables.
+		verifyLevelMetadataTransition(&o.levelMetadata, newLevelMeta, addedL0Tables, deletedL0Tables)
+	}
+	o.levelMetadata = *newLevelMeta
+	if len(addedL0Tables) == 0 && len(deletedL0Tables) == 0 {
+		return
+	}
+	// If we only added tables, try to use AddL0Files.
+	if len(deletedL0Tables) == 0 {
+		// Construct the file slice needed by AddL0Files.
+		// TODO(radu): change AddL0Files to do this internally.
+		files := make([]*TableMetadata, 0, len(addedL0Tables))
+		iter := newLevelMeta.Iter()
+		for t := iter.Last(); len(files) < len(addedL0Tables); t = iter.Prev() {
+			if t == nil || addedL0Tables[t.FileNum] == nil {
+				break
+			}
+			files = append(files, t)
+		}
+		if len(files) == len(addedL0Tables) {
+			slices.Reverse(files)
+			newSublevels, err := o.sublevels.AddL0Files(files, o.flushSplitBytes, newLevelMeta)
+			if err == nil {
+				// In invariants mode, sometimes rebuild from scratch to verify that
+				// AddL0Files did the right thing. Note that NewL0Sublevels updates
+				// fields in TableMetadata like L0Index, so we don't want to do this
+				// every time.
+				if invariants.Enabled && invariants.Sometimes(10) {
+					expectedSublevels, err := NewL0Sublevels(newLevelMeta, o.cmp, o.formatKey, o.flushSplitBytes)
+					if err != nil {
+						panic(fmt.Sprintf("error when regenerating sublevels: %s", err))
+					}
+					s1 := describeSublevels(o.formatKey, false /* verbose */, expectedSublevels.Levels)
+					s2 := describeSublevels(o.formatKey, false /* verbose */, newSublevels.Levels)
+					if s1 != s2 {
+						// Add verbosity.
+						s1 := describeSublevels(o.formatKey, true /* verbose */, expectedSublevels.Levels)
+						s2 := describeSublevels(o.formatKey, true /* verbose */, newSublevels.Levels)
+						panic(fmt.Sprintf("incremental L0 sublevel generation produced different output than regeneration: %s != %s", s1, s2))
+					}
+				}
+				o.sublevels = newSublevels
+				return
+			}
+			if !errors.Is(err, errInvalidL0SublevelsOpt) {
+				panic(errors.AssertionFailedf("error generating L0Sublevels: %s", err))
+			}
+		}
+	}
+	var err error
+	o.sublevels, err = NewL0Sublevels(newLevelMeta, o.cmp, o.formatKey, o.flushSplitBytes)
+	if err != nil {
+		panic(errors.AssertionFailedf("error generating L0Sublevels: %s", err))
+	}
+}
+
+// Reset the L0Organizer to reflect a given L0 level. Used for testing.
+func (o *L0Organizer) Reset(levelMetadata *LevelMetadata) {
+	o.levelMetadata = *levelMetadata
+	var err error
+	o.sublevels, err = NewL0Sublevels(levelMetadata, o.cmp, o.formatKey, o.flushSplitBytes)
+	if err != nil {
+		panic(errors.AssertionFailedf("error generating L0Sublevels: %s", err))
+	}
+}
+
+// verifyLevelMetadataTransition verifies that newLevel matches oldLevel after
+// adding and removing the specified tables.
+func verifyLevelMetadataTransition(
+	oldLevel, newLevel *LevelMetadata,
+	addedTables map[base.FileNum]*TableMetadata,
+	deletedTables map[base.FileNum]*TableMetadata,
+) {
+	m := make(map[base.FileNum]*TableMetadata, oldLevel.Len())
+	iter := oldLevel.Iter()
+	for t := iter.First(); t != nil; t = iter.Next() {
+		m[t.FileNum] = t
+	}
+	for n, t := range addedTables {
+		if m[n] != nil {
+			panic("added table that already exists in old level")
+		}
+		m[n] = t
+	}
+	for n, t := range deletedTables {
+		if m[n] == nil {
+			panic("deleted table not in old level")
+		}
+		if m[n] != t {
+			panic("deleted table does not match old level")
+		}
+		delete(m, n)
+	}
+	iter = newLevel.Iter()
+	for t := iter.First(); t != nil; t = iter.Next() {
+		if m[t.FileNum] == nil {
+			panic("unknown table in new level")
+		}
+		delete(m, t.FileNum)
+	}
+	if len(m) != 0 {
+		panic("tables missing from the new level")
+	}
+}

internal/manifest/l0_sublevels_test.go

@@ -32,7 +32,9 @@ func readManifest(filename string) (*Version, error) {
 	}
 	defer f.Close()
 	rr := record.NewReader(f, 0 /* logNum */)
-	v := NewVersion(base.DefaultComparer)
+
+	l0Organizer := NewL0Organizer(base.DefaultComparer, 10<<20 /* flushSplitBytes */)
+	v := NewInitialVersion(base.DefaultComparer, l0Organizer)
 	addedByFileNum := make(map[base.FileNum]*TableMetadata)
 	for {
 		r, err := rr.Next()
@@ -51,7 +53,7 @@ func readManifest(filename string) (*Version, error) {
 		if err := bve.Accumulate(&ve); err != nil {
 			return nil, err
 		}
-		if v, err = bve.Apply(v, 10<<20, 32000); err != nil {
+		if v, err = bve.Apply(v, l0Organizer, 32000); err != nil {
 			return nil, err
 		}
 	}
@@ -318,7 +320,7 @@ func TestL0Sublevels(t *testing.T) {
 					levelFiles:    explicitSublevels,
 					cmp:           base.DefaultComparer.Compare,
 					formatKey:     base.DefaultFormatter,
-					levelMetadata: &levelMetadata,
+					levelMetadata: levelMetadata,
 				}
 				for _, files := range explicitSublevels {
 					sublevels.Levels = append(sublevels.Levels, NewLevelSliceSpecificOrder(files))