Line data Source code
1 : // Copyright 2012 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package pebble
6 :
7 : import (
8 : "fmt"
9 : "io"
10 : "sync"
11 : "sync/atomic"
12 :
13 : "github.com/cockroachdb/errors"
14 : "github.com/cockroachdb/pebble/internal/base"
15 : "github.com/cockroachdb/pebble/internal/invariants"
16 : "github.com/cockroachdb/pebble/internal/manifest"
17 : "github.com/cockroachdb/pebble/objstorage"
18 : "github.com/cockroachdb/pebble/record"
19 : "github.com/cockroachdb/pebble/vfs"
20 : "github.com/cockroachdb/pebble/vfs/atomicfs"
21 : )
22 :
23 : const numLevels = manifest.NumLevels
24 :
25 : const manifestMarkerName = `manifest`
26 :
27 : // Provide type aliases for the various manifest structs.
28 : type bulkVersionEdit = manifest.BulkVersionEdit
29 : type tableMetadata = manifest.TableMetadata
30 : type fileBacking = manifest.FileBacking
31 : type newTableEntry = manifest.NewTableEntry
32 : type version = manifest.Version
33 : type versionEdit = manifest.VersionEdit
34 : type versionList = manifest.VersionList
35 :
36 : // versionSet manages a collection of immutable versions, and manages the
37 : // creation of a new version from the most recent version. A new version is
38 : // created from an existing version by applying a version edit which is just
39 : // like it sounds: a delta from the previous version. Version edits are logged
40 : // to the MANIFEST file, which is replayed at startup.
41 : type versionSet struct {
42 : // Next seqNum to use for WAL writes.
43 : logSeqNum base.AtomicSeqNum
44 :
45 : // The upper bound on sequence numbers that have been assigned so far. A
46 : // suffix of these sequence numbers may not have been written to a WAL. Both
47 : // logSeqNum and visibleSeqNum are atomically updated by the commitPipeline.
48 : // visibleSeqNum is <= logSeqNum.
49 : visibleSeqNum base.AtomicSeqNum
50 :
51 : // Number of bytes present in sstables being written by in-progress
52 : // compactions. This value will be zero if there are no in-progress
53 : // compactions. Updated and read atomically.
54 : atomicInProgressBytes atomic.Int64
55 :
56 : // Immutable fields.
57 : dirname string
58 : provider objstorage.Provider
59 : // Set to DB.mu.
60 : mu *sync.Mutex
61 : opts *Options
62 : fs vfs.FS
63 : cmp *base.Comparer
64 : // Dynamic base level allows the dynamic base level computation to be
65 : // disabled. Used by tests which want to create specific LSM structures.
66 : dynamicBaseLevel bool
67 :
68 : // Mutable fields.
69 : versions versionList
70 : l0Organizer *manifest.L0Organizer
71 : // blobFiles is the set of blob files referenced by the current version.
72 : // blobFiles is protected by the manifest logLock (not vs.mu).
73 : blobFiles manifest.CurrentBlobFileSet
74 : picker compactionPicker
75 : // curCompactionConcurrency is updated whenever picker is updated.
76 : // INVARIANT: >= 1.
77 : curCompactionConcurrency atomic.Int32
78 :
79 : // Not all metrics are kept here. See DB.Metrics().
80 : metrics Metrics
81 :
82 : // A pointer to versionSet.addObsoleteLocked. Avoids allocating a new closure
83 : // on the creation of every version.
84 : obsoleteFn func(manifest.ObsoleteFiles)
85 : // obsolete{Tables,Blobs,Manifests,Options} are sorted by file number ascending.
86 : obsoleteTables []objectInfo
87 : obsoleteBlobs []objectInfo
88 : obsoleteManifests []obsoleteFile
89 : obsoleteOptions []obsoleteFile
90 :
91 : // Zombie tables which have been removed from the current version but are
92 : // still referenced by an inuse iterator.
93 : zombieTables zombieObjects
94 : // Zombie blobs which have been removed from the current version but are
95 : // still referenced by an inuse iterator.
96 : zombieBlobs zombieObjects
97 : // virtualBackings contains information about the FileBackings which support
98 : // virtual sstables in the latest version. It is mainly used to determine when
99 : // a backing is no longer in use by the tables in the latest version; this is
100 : // not a trivial problem because compactions and other operations can happen
101 : // in parallel (and they can finish in unpredictable order).
102 : //
103 : // This mechanism is complementary to the backing Ref/Unref mechanism, which
104 : // determines when a backing is no longer used by *any* live version and can
105 : // be removed.
106 : //
107 : // In principle this should have been a copy-on-write structure, with each
108 : // Version having its own record of its virtual backings (similar to the
109 : // B-tree that holds the tables). However, in practice we only need it for the
110 : // latest version, so we use a simpler structure and store it in the
111 : // versionSet instead.
112 : //
113 : // virtualBackings is modified under DB.mu and the log lock. If it is accessed
114 : // under DB.mu and a version update is in progress, it reflects the state of
115 : // the next version.
116 : virtualBackings manifest.VirtualBackings
117 :
118 : // minUnflushedLogNum is the smallest WAL log file number corresponding to
119 : // mutations that have not been flushed to an sstable.
120 : minUnflushedLogNum base.DiskFileNum
121 :
122 : // The next file number. A single counter is used to assign file
123 : // numbers for the WAL, MANIFEST, sstable, and OPTIONS files.
124 : nextFileNum atomic.Uint64
125 :
126 : // The current manifest file number.
127 : manifestFileNum base.DiskFileNum
128 : manifestMarker *atomicfs.Marker
129 :
130 : manifestFile vfs.File
131 : manifest *record.Writer
132 : getFormatMajorVersion func() FormatMajorVersion
133 :
134 : writing bool
135 : writerCond sync.Cond
136 : // State for deciding when to write a snapshot. Protected by mu.
137 : rotationHelper record.RotationHelper
138 :
139 : pickedCompactionCache pickedCompactionCache
140 : }
141 :
142 : func (vs *versionSet) init(
143 : dirname string,
144 : provider objstorage.Provider,
145 : opts *Options,
146 : marker *atomicfs.Marker,
147 : getFMV func() FormatMajorVersion,
148 : mu *sync.Mutex,
149 1 : ) {
150 1 : vs.dirname = dirname
151 1 : vs.provider = provider
152 1 : vs.mu = mu
153 1 : vs.writerCond.L = mu
154 1 : vs.opts = opts
155 1 : vs.fs = opts.FS
156 1 : vs.cmp = opts.Comparer
157 1 : vs.dynamicBaseLevel = true
158 1 : vs.versions.Init(mu)
159 1 : vs.l0Organizer = manifest.NewL0Organizer(opts.Comparer, opts.FlushSplitBytes)
160 1 : vs.obsoleteFn = vs.addObsoleteLocked
161 1 : vs.zombieTables = makeZombieObjects()
162 1 : vs.zombieBlobs = makeZombieObjects()
163 1 : vs.virtualBackings = manifest.MakeVirtualBackings()
164 1 : vs.nextFileNum.Store(1)
165 1 : vs.manifestMarker = marker
166 1 : vs.getFormatMajorVersion = getFMV
167 1 : }
168 :
169 : // create creates a version set for a fresh DB.
170 : func (vs *versionSet) create(
171 : jobID JobID,
172 : dirname string,
173 : provider objstorage.Provider,
174 : opts *Options,
175 : marker *atomicfs.Marker,
176 : getFormatMajorVersion func() FormatMajorVersion,
177 : mu *sync.Mutex,
178 1 : ) error {
179 1 : vs.init(dirname, provider, opts, marker, getFormatMajorVersion, mu)
180 1 : emptyVersion := manifest.NewInitialVersion(opts.Comparer)
181 1 : vs.append(emptyVersion)
182 1 : vs.blobFiles.Init(nil)
183 1 :
184 1 : vs.setCompactionPicker(
185 1 : newCompactionPickerByScore(emptyVersion, vs.l0Organizer, &vs.virtualBackings, vs.opts, nil))
186 1 : // Note that a "snapshot" version edit is written to the manifest when it is
187 1 : // created.
188 1 : vs.manifestFileNum = vs.getNextDiskFileNum()
189 1 : err := vs.createManifest(vs.dirname, vs.manifestFileNum, vs.minUnflushedLogNum, vs.nextFileNum.Load(), nil /* virtualBackings */)
190 1 : if err == nil {
191 1 : if err = vs.manifest.Flush(); err != nil {
192 1 : vs.opts.Logger.Fatalf("MANIFEST flush failed: %v", err)
193 1 : }
194 : }
195 1 : if err == nil {
196 1 : if err = vs.manifestFile.Sync(); err != nil {
197 1 : vs.opts.Logger.Fatalf("MANIFEST sync failed: %v", err)
198 1 : }
199 : }
200 1 : if err == nil {
201 1 : // NB: Move() is responsible for syncing the data directory.
202 1 : if err = vs.manifestMarker.Move(base.MakeFilename(base.FileTypeManifest, vs.manifestFileNum)); err != nil {
203 1 : vs.opts.Logger.Fatalf("MANIFEST set current failed: %v", err)
204 1 : }
205 : }
206 :
207 1 : vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
208 1 : JobID: int(jobID),
209 1 : Path: base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, vs.manifestFileNum),
210 1 : FileNum: vs.manifestFileNum,
211 1 : Err: err,
212 1 : })
213 1 : if err != nil {
214 1 : return err
215 1 : }
216 1 : return nil
217 : }
218 :
219 : // load loads the version set from the manifest file.
220 : func (vs *versionSet) load(
221 : dirname string,
222 : provider objstorage.Provider,
223 : opts *Options,
224 : manifestFileNum base.DiskFileNum,
225 : marker *atomicfs.Marker,
226 : getFormatMajorVersion func() FormatMajorVersion,
227 : mu *sync.Mutex,
228 1 : ) error {
229 1 : vs.init(dirname, provider, opts, marker, getFormatMajorVersion, mu)
230 1 :
231 1 : vs.manifestFileNum = manifestFileNum
232 1 : manifestPath := base.MakeFilepath(opts.FS, dirname, base.FileTypeManifest, vs.manifestFileNum)
233 1 : manifestFilename := opts.FS.PathBase(manifestPath)
234 1 :
235 1 : // Read the versionEdits in the manifest file.
236 1 : var bve bulkVersionEdit
237 1 : bve.AllAddedTables = make(map[base.FileNum]*tableMetadata)
238 1 : manifestFile, err := vs.fs.Open(manifestPath)
239 1 : if err != nil {
240 0 : return errors.Wrapf(err, "pebble: could not open manifest file %q for DB %q",
241 0 : errors.Safe(manifestFilename), dirname)
242 0 : }
243 1 : defer manifestFile.Close()
244 1 : rr := record.NewReader(manifestFile, 0 /* logNum */)
245 1 : for {
246 1 : r, err := rr.Next()
247 1 : if err == io.EOF || record.IsInvalidRecord(err) {
248 1 : break
249 : }
250 1 : if err != nil {
251 0 : return errors.Wrapf(err, "pebble: error when loading manifest file %q",
252 0 : errors.Safe(manifestFilename))
253 0 : }
254 1 : var ve versionEdit
255 1 : err = ve.Decode(r)
256 1 : if err != nil {
257 1 : // Break instead of returning an error if the record is corrupted
258 1 : // or invalid.
259 1 : if err == io.EOF || record.IsInvalidRecord(err) {
260 1 : break
261 : }
262 0 : return err
263 : }
264 1 : if ve.ComparerName != "" {
265 1 : if ve.ComparerName != vs.cmp.Name {
266 1 : return errors.Errorf("pebble: manifest file %q for DB %q: "+
267 1 : "comparer name from file %q != comparer name from Options %q",
268 1 : errors.Safe(manifestFilename), dirname, errors.Safe(ve.ComparerName), errors.Safe(vs.cmp.Name))
269 1 : }
270 : }
271 1 : if err := bve.Accumulate(&ve); err != nil {
272 0 : return err
273 0 : }
274 1 : if ve.MinUnflushedLogNum != 0 {
275 1 : vs.minUnflushedLogNum = ve.MinUnflushedLogNum
276 1 : }
277 1 : if ve.NextFileNum != 0 {
278 1 : vs.nextFileNum.Store(ve.NextFileNum)
279 1 : }
280 1 : if ve.LastSeqNum != 0 {
281 1 : // logSeqNum is the _next_ sequence number that will be assigned,
282 1 : // while LastSeqNum is the last assigned sequence number. Note that
283 1 : // this behaviour mimics that in RocksDB; the first sequence number
284 1 : // assigned is one greater than the one present in the manifest
285 1 : // (assuming no WALs contain higher sequence numbers than the
286 1 : // manifest's LastSeqNum). Increment LastSeqNum by 1 to get the
287 1 : // next sequence number that will be assigned.
288 1 : //
289 1 : // If LastSeqNum is less than SeqNumStart, increase it to at least
290 1 : // SeqNumStart to leave ample room for reserved sequence numbers.
291 1 : if ve.LastSeqNum+1 < base.SeqNumStart {
292 0 : vs.logSeqNum.Store(base.SeqNumStart)
293 1 : } else {
294 1 : vs.logSeqNum.Store(ve.LastSeqNum + 1)
295 1 : }
296 : }
297 : }
298 : // We have already set vs.nextFileNum = 2 at the beginning of the
299 : // function and could have only updated it to some other non-zero value,
300 : // so it cannot be 0 here.
301 1 : if vs.minUnflushedLogNum == 0 {
302 1 : if vs.nextFileNum.Load() >= 2 {
303 1 : // We either have a freshly created DB, or a DB created by RocksDB
304 1 : // that has not had a single flushed SSTable yet. This is because
305 1 : // RocksDB bumps up nextFileNum in this case without bumping up
306 1 : // minUnflushedLogNum, even if WALs with non-zero file numbers are
307 1 : // present in the directory.
308 1 : } else {
309 0 : return base.CorruptionErrorf("pebble: malformed manifest file %q for DB %q",
310 0 : errors.Safe(manifestFilename), dirname)
311 0 : }
312 : }
313 1 : vs.markFileNumUsed(vs.minUnflushedLogNum)
314 1 :
315 1 : // Populate the fileBackingMap and the FileBacking for virtual sstables since
316 1 : // we have finished version edit accumulation.
317 1 : for _, b := range bve.AddedFileBacking {
318 1 : vs.virtualBackings.AddAndRef(b)
319 1 : }
320 :
321 1 : for _, addedLevel := range bve.AddedTables {
322 1 : for _, m := range addedLevel {
323 1 : if m.Virtual {
324 1 : vs.virtualBackings.AddTable(m)
325 1 : }
326 : }
327 : }
328 :
329 1 : if invariants.Enabled {
330 1 : // There should be no deleted tables or backings, since we're starting from
331 1 : // an empty state.
332 1 : for _, deletedLevel := range bve.DeletedTables {
333 1 : if len(deletedLevel) != 0 {
334 0 : panic("deleted files after manifest replay")
335 : }
336 : }
337 1 : if len(bve.RemovedFileBacking) > 0 {
338 0 : panic("deleted backings after manifest replay")
339 : }
340 : }
341 :
342 1 : emptyVersion := manifest.NewInitialVersion(opts.Comparer)
343 1 : newVersion, err := bve.Apply(emptyVersion, opts.Experimental.ReadCompactionRate)
344 1 : if err != nil {
345 0 : return err
346 0 : }
347 1 : vs.l0Organizer.PerformUpdate(vs.l0Organizer.PrepareUpdate(&bve, newVersion), newVersion)
348 1 : vs.l0Organizer.InitCompactingFileInfo(nil /* in-progress compactions */)
349 1 : vs.blobFiles.Init(&bve)
350 1 : vs.append(newVersion)
351 1 :
352 1 : for i := range vs.metrics.Levels {
353 1 : l := &vs.metrics.Levels[i]
354 1 : l.TablesCount = int64(newVersion.Levels[i].Len())
355 1 : files := newVersion.Levels[i].Slice()
356 1 : l.TablesSize = int64(files.TableSizeSum())
357 1 : }
358 1 : for _, l := range newVersion.Levels {
359 1 : for f := range l.All() {
360 1 : if !f.Virtual {
361 1 : isLocal, localSize := sizeIfLocal(f.FileBacking, vs.provider)
362 1 : vs.metrics.Table.Local.LiveSize = uint64(int64(vs.metrics.Table.Local.LiveSize) + localSize)
363 1 : if isLocal {
364 1 : vs.metrics.Table.Local.LiveCount++
365 1 : }
366 : }
367 : }
368 : }
369 1 : vs.virtualBackings.ForEach(func(backing *fileBacking) {
370 1 : isLocal, localSize := sizeIfLocal(backing, vs.provider)
371 1 : vs.metrics.Table.Local.LiveSize = uint64(int64(vs.metrics.Table.Local.LiveSize) + localSize)
372 1 : if isLocal {
373 1 : vs.metrics.Table.Local.LiveCount++
374 1 : }
375 : })
376 :
377 1 : vs.setCompactionPicker(
378 1 : newCompactionPickerByScore(newVersion, vs.l0Organizer, &vs.virtualBackings, vs.opts, nil))
379 1 : return nil
380 : }
381 :
382 1 : func (vs *versionSet) close() error {
383 1 : if vs.manifestFile != nil {
384 1 : if err := vs.manifestFile.Close(); err != nil {
385 0 : return err
386 0 : }
387 : }
388 1 : if vs.manifestMarker != nil {
389 1 : if err := vs.manifestMarker.Close(); err != nil {
390 0 : return err
391 0 : }
392 : }
393 1 : return nil
394 : }
395 :
396 : // logLock locks the manifest for writing. The lock must be released by
397 : // a call to logUnlock.
398 : //
399 : // DB.mu must be held when calling this method, but the mutex may be dropped and
400 : // re-acquired during the course of this method.
401 1 : func (vs *versionSet) logLock() {
402 1 : // Wait for any existing writing to the manifest to complete, then mark the
403 1 : // manifest as busy.
404 1 : for vs.writing {
405 1 : // Note: writerCond.L is DB.mu, so we unlock it while we wait.
406 1 : vs.writerCond.Wait()
407 1 : }
408 1 : vs.writing = true
409 : }
410 :
411 : // logUnlock releases the lock for manifest writing.
412 : //
413 : // DB.mu must be held when calling this method.
414 1 : func (vs *versionSet) logUnlock() {
415 1 : if !vs.writing {
416 0 : vs.opts.Logger.Fatalf("MANIFEST not locked for writing")
417 0 : }
418 1 : vs.writing = false
419 1 : vs.writerCond.Signal()
420 : }
421 :
422 1 : func (vs *versionSet) logUnlockAndInvalidatePickedCompactionCache() {
423 1 : vs.pickedCompactionCache.invalidate()
424 1 : vs.logUnlock()
425 1 : }
426 :
427 : // versionUpdate is returned by the function passed to UpdateVersionLocked.
428 : //
429 : // If VE is nil, there is no update to apply (but it is not an error).
430 : type versionUpdate struct {
431 : VE *manifest.VersionEdit
432 : JobID JobID
433 : Metrics levelMetricsDelta
434 : // InProgressCompactionFn is called while DB.mu is held after the I/O part of
435 : // the update was performed. It should return any compactions that are
436 : // in-progress (excluding than the one that is being applied).
437 : InProgressCompactionsFn func() []compactionInfo
438 : ForceManifestRotation bool
439 : }
440 :
441 : // UpdateVersionLocked is used to update the current version.
442 : //
443 : // DB.mu must be held. UpdateVersionLocked first waits for any other version
444 : // update to complete, releasing and reacquiring DB.mu.
445 : //
446 : // UpdateVersionLocked then calls updateFn which builds a versionUpdate, while
447 : // holding DB.mu. The updateFn can release and reacquire DB.mu (it should
448 : // attempt to do as much work as possible outside of the lock).
449 : //
450 : // UpdateVersionLocked fills in the following fields of the VersionEdit:
451 : // NextFileNum, LastSeqNum, RemovedBackingTables. The removed backing tables are
452 : // those backings that are no longer used (in the new version) after applying
453 : // the edit (as per vs.virtualBackings). Other than these fields, the
454 : // VersionEdit must be complete.
455 : //
456 : // New table backing references (FileBacking.Ref) are taken as part of applying
457 : // the version edit. The state of the virtual backings (vs.virtualBackings) is
458 : // updated before logging to the manifest and installing the latest version;
459 : // this is ok because any failure in those steps is fatal.
460 : //
461 : // If updateFn returns an error, no update is applied and that same error is returned.
462 : // If versionUpdate.VE is nil, the no update is applied (and no error is returned).
463 1 : func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)) error {
464 1 : vs.logLock()
465 1 : defer vs.logUnlockAndInvalidatePickedCompactionCache()
466 1 :
467 1 : vu, err := updateFn()
468 1 : if err != nil || vu.VE == nil {
469 1 : return err
470 1 : }
471 :
472 1 : if !vs.writing {
473 0 : vs.opts.Logger.Fatalf("MANIFEST not locked for writing")
474 0 : }
475 :
476 1 : ve := vu.VE
477 1 : if ve.MinUnflushedLogNum != 0 {
478 1 : if ve.MinUnflushedLogNum < vs.minUnflushedLogNum ||
479 1 : vs.nextFileNum.Load() <= uint64(ve.MinUnflushedLogNum) {
480 0 : panic(fmt.Sprintf("pebble: inconsistent versionEdit minUnflushedLogNum %d",
481 0 : ve.MinUnflushedLogNum))
482 : }
483 : }
484 :
485 : // This is the next manifest filenum, but if the current file is too big we
486 : // will write this ve to the next file which means what ve encodes is the
487 : // current filenum and not the next one.
488 : //
489 : // TODO(sbhola): figure out why this is correct and update comment.
490 1 : ve.NextFileNum = vs.nextFileNum.Load()
491 1 :
492 1 : // LastSeqNum is set to the current upper bound on the assigned sequence
493 1 : // numbers. Note that this is exactly the behavior of RocksDB. LastSeqNum is
494 1 : // used to initialize versionSet.logSeqNum and versionSet.visibleSeqNum on
495 1 : // replay. It must be higher than or equal to any than any sequence number
496 1 : // written to an sstable, including sequence numbers in ingested files.
497 1 : // Note that LastSeqNum is not (and cannot be) the minimum unflushed sequence
498 1 : // number. This is fallout from ingestion which allows a sequence number X to
499 1 : // be assigned to an ingested sstable even though sequence number X-1 resides
500 1 : // in an unflushed memtable. logSeqNum is the _next_ sequence number that
501 1 : // will be assigned, so subtract that by 1 to get the upper bound on the
502 1 : // last assigned sequence number.
503 1 : logSeqNum := vs.logSeqNum.Load()
504 1 : ve.LastSeqNum = logSeqNum - 1
505 1 : if logSeqNum == 0 {
506 0 : // logSeqNum is initialized to 1 in Open() if there are no previous WAL
507 0 : // or manifest records, so this case should never happen.
508 0 : vs.opts.Logger.Fatalf("logSeqNum must be a positive integer: %d", logSeqNum)
509 0 : }
510 :
511 1 : currentVersion := vs.currentVersion()
512 1 : var newVersion *version
513 1 :
514 1 : // Generate a new manifest if we don't currently have one, or forceRotation
515 1 : // is true, or the current one is too large.
516 1 : //
517 1 : // For largeness, we do not exclusively use MaxManifestFileSize size
518 1 : // threshold since we have had incidents where due to either large keys or
519 1 : // large numbers of files, each edit results in a snapshot + write of the
520 1 : // edit. This slows the system down since each flush or compaction is
521 1 : // writing a new manifest snapshot. The primary goal of the size-based
522 1 : // rollover logic is to ensure that when reopening a DB, the number of edits
523 1 : // that need to be replayed on top of the snapshot is "sane". Rolling over
524 1 : // to a new manifest after each edit is not relevant to that goal.
525 1 : //
526 1 : // Consider the following cases:
527 1 : // - The number of live files F in the DB is roughly stable: after writing
528 1 : // the snapshot (with F files), say we require that there be enough edits
529 1 : // such that the cumulative number of files in those edits, E, be greater
530 1 : // than F. This will ensure that the total amount of time in
531 1 : // UpdateVersionLocked that is spent in snapshot writing is ~50%.
532 1 : //
533 1 : // - The number of live files F in the DB is shrinking drastically, say from
534 1 : // F to F/10: This can happen for various reasons, like wide range
535 1 : // tombstones, or large numbers of smaller than usual files that are being
536 1 : // merged together into larger files. And say the new files generated
537 1 : // during this shrinkage is insignificant compared to F/10, and so for
538 1 : // this example we will assume it is effectively 0. After this shrinking,
539 1 : // E = 0.9F, and so if we used the previous snapshot file count, F, as the
540 1 : // threshold that needs to be exceeded, we will further delay the snapshot
541 1 : // writing. Which means on DB reopen we will need to replay 0.9F edits to
542 1 : // get to a version with 0.1F files. It would be better to create a new
543 1 : // snapshot when E exceeds the number of files in the current version.
544 1 : //
545 1 : // - The number of live files F in the DB is growing via perfect ingests
546 1 : // into L6: Say we wrote the snapshot when there were F files and now we
547 1 : // have 10F files, so E = 9F. We will further delay writing a new
548 1 : // snapshot. This case can be critiqued as contrived, but we consider it
549 1 : // nonetheless.
550 1 : //
551 1 : // The logic below uses the min of the last snapshot file count and the file
552 1 : // count in the current version.
553 1 : vs.rotationHelper.AddRecord(int64(len(ve.DeletedTables) + len(ve.NewTables)))
554 1 : sizeExceeded := vs.manifest.Size() >= vs.opts.MaxManifestFileSize
555 1 : requireRotation := vu.ForceManifestRotation || vs.manifest == nil
556 1 :
557 1 : var nextSnapshotFilecount int64
558 1 : for i := range vs.metrics.Levels {
559 1 : nextSnapshotFilecount += vs.metrics.Levels[i].TablesCount
560 1 : }
561 1 : if sizeExceeded && !requireRotation {
562 1 : requireRotation = vs.rotationHelper.ShouldRotate(nextSnapshotFilecount)
563 1 : }
564 1 : var newManifestFileNum base.DiskFileNum
565 1 : var prevManifestFileSize uint64
566 1 : var newManifestVirtualBackings []*fileBacking
567 1 : if requireRotation {
568 1 : newManifestFileNum = vs.getNextDiskFileNum()
569 1 : prevManifestFileSize = uint64(vs.manifest.Size())
570 1 :
571 1 : // We want the virtual backings *before* applying the version edit, because
572 1 : // the new manifest will contain the pre-apply version plus the last version
573 1 : // edit.
574 1 : newManifestVirtualBackings = vs.virtualBackings.Backings()
575 1 : }
576 :
577 : // Grab certain values before releasing vs.mu, in case createManifest() needs
578 : // to be called.
579 1 : minUnflushedLogNum := vs.minUnflushedLogNum
580 1 : nextFileNum := vs.nextFileNum.Load()
581 1 :
582 1 : // Note: this call populates ve.RemovedBackingTables.
583 1 : zombieBackings, removedVirtualBackings, localTablesLiveDelta :=
584 1 : getZombieTablesAndUpdateVirtualBackings(ve, &vs.virtualBackings, vs.provider)
585 1 :
586 1 : var l0Update manifest.L0PreparedUpdate
587 1 : if err := func() error {
588 1 : vs.mu.Unlock()
589 1 : defer vs.mu.Lock()
590 1 :
591 1 : if vs.getFormatMajorVersion() < FormatVirtualSSTables && len(ve.CreatedBackingTables) > 0 {
592 0 : return base.AssertionFailedf("MANIFEST cannot contain virtual sstable records due to format major version")
593 0 : }
594 :
595 : // Rotate the manifest if necessary. Rotating the manifest involves
596 : // creating a new file and writing an initial version edit containing a
597 : // snapshot of the current version. This initial version edit will
598 : // reflect the Version prior to the pending version edit (`ve`). Once
599 : // we've created the new manifest with the previous version state, we'll
600 : // append the version edit `ve` to the tail of the new manifest.
601 1 : if newManifestFileNum != 0 {
602 1 : if err := vs.createManifest(vs.dirname, newManifestFileNum, minUnflushedLogNum, nextFileNum, newManifestVirtualBackings); err != nil {
603 1 : vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
604 1 : JobID: int(vu.JobID),
605 1 : Path: base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, newManifestFileNum),
606 1 : FileNum: newManifestFileNum,
607 1 : Err: err,
608 1 : })
609 1 : return errors.Wrap(err, "MANIFEST create failed")
610 1 : }
611 : }
612 :
613 : // Call ApplyAndUpdateVersionEdit before accumulating the version edit.
614 : // If any blob files are no longer referenced, the version edit will be
615 : // updated to explicitly record the deletion of the blob files. This can
616 : // happen here because vs.blobFiles is protected by the manifest logLock
617 : // (NOT vs.mu). We only read or write vs.blobFiles while holding the
618 : // manifest lock.
619 1 : if err := vs.blobFiles.ApplyAndUpdateVersionEdit(ve); err != nil {
620 0 : return errors.Wrap(err, "MANIFEST blob files apply and update failed")
621 0 : }
622 :
623 1 : var bulkEdit bulkVersionEdit
624 1 : err := bulkEdit.Accumulate(ve)
625 1 : if err != nil {
626 0 : return errors.Wrap(err, "MANIFEST accumulate failed")
627 0 : }
628 1 : newVersion, err = bulkEdit.Apply(currentVersion, vs.opts.Experimental.ReadCompactionRate)
629 1 : if err != nil {
630 0 : return errors.Wrap(err, "MANIFEST apply failed")
631 0 : }
632 1 : l0Update = vs.l0Organizer.PrepareUpdate(&bulkEdit, newVersion)
633 1 :
634 1 : w, err := vs.manifest.Next()
635 1 : if err != nil {
636 0 : return errors.Wrap(err, "MANIFEST next record write failed")
637 0 : }
638 :
639 : // NB: Any error from this point on is considered fatal as we don't know if
640 : // the MANIFEST write occurred or not. Trying to determine that is
641 : // fraught. Instead we rely on the standard recovery mechanism run when a
642 : // database is open. In particular, that mechanism generates a new MANIFEST
643 : // and ensures it is synced.
644 1 : if err := ve.Encode(w); err != nil {
645 0 : return errors.Wrap(err, "MANIFEST write failed")
646 0 : }
647 1 : if err := vs.manifest.Flush(); err != nil {
648 1 : return errors.Wrap(err, "MANIFEST flush failed")
649 1 : }
650 1 : if err := vs.manifestFile.Sync(); err != nil {
651 1 : return errors.Wrap(err, "MANIFEST sync failed")
652 1 : }
653 1 : if newManifestFileNum != 0 {
654 1 : // NB: Move() is responsible for syncing the data directory.
655 1 : if err := vs.manifestMarker.Move(base.MakeFilename(base.FileTypeManifest, newManifestFileNum)); err != nil {
656 0 : return errors.Wrap(err, "MANIFEST set current failed")
657 0 : }
658 1 : vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
659 1 : JobID: int(vu.JobID),
660 1 : Path: base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, newManifestFileNum),
661 1 : FileNum: newManifestFileNum,
662 1 : })
663 : }
664 1 : return nil
665 1 : }(); err != nil {
666 1 : // Any error encountered during any of the operations in the previous
667 1 : // closure are considered fatal. Treating such errors as fatal is preferred
668 1 : // to attempting to unwind various file and b-tree reference counts, and
669 1 : // re-generating L0 sublevel metadata. This may change in the future, if
670 1 : // certain manifest / WAL operations become retryable. For more context, see
671 1 : // #1159 and #1792.
672 1 : vs.opts.Logger.Fatalf("%s", err)
673 1 : return err
674 1 : }
675 :
676 1 : if requireRotation {
677 1 : // Successfully rotated.
678 1 : vs.rotationHelper.Rotate(nextSnapshotFilecount)
679 1 : }
680 : // Now that DB.mu is held again, initialize compacting file info in
681 : // L0Sublevels.
682 1 : inProgress := vu.InProgressCompactionsFn()
683 1 :
684 1 : zombieBlobs, localBlobLiveDelta := getZombieBlobFilesAndComputeLocalMetrics(ve, vs.provider)
685 1 : vs.l0Organizer.PerformUpdate(l0Update, newVersion)
686 1 : vs.l0Organizer.InitCompactingFileInfo(inProgressL0Compactions(inProgress))
687 1 :
688 1 : // Update the zombie objects sets first, as installation of the new version
689 1 : // will unref the previous version which could result in addObsoleteLocked
690 1 : // being called.
691 1 : for _, b := range zombieBackings {
692 1 : vs.zombieTables.Add(objectInfo{
693 1 : fileInfo: fileInfo{
694 1 : FileNum: b.backing.DiskFileNum,
695 1 : FileSize: b.backing.Size,
696 1 : },
697 1 : isLocal: b.isLocal,
698 1 : })
699 1 : }
700 1 : for _, zb := range zombieBlobs {
701 1 : vs.zombieBlobs.Add(zb)
702 1 : }
703 : // Unref the removed backings and report those that already became obsolete.
704 : // Note that the only case where we report obsolete tables here is when
705 : // VirtualBackings.Protect/Unprotect was used to keep a backing alive without
706 : // it being used in the current version.
707 1 : var obsoleteVirtualBackings manifest.ObsoleteFiles
708 1 : for _, b := range removedVirtualBackings {
709 1 : if b.backing.Unref() == 0 {
710 1 : obsoleteVirtualBackings.FileBackings = append(obsoleteVirtualBackings.FileBackings, b.backing)
711 1 : }
712 : }
713 1 : vs.addObsoleteLocked(obsoleteVirtualBackings)
714 1 :
715 1 : // Install the new version.
716 1 : vs.append(newVersion)
717 1 :
718 1 : if ve.MinUnflushedLogNum != 0 {
719 1 : vs.minUnflushedLogNum = ve.MinUnflushedLogNum
720 1 : }
721 1 : if newManifestFileNum != 0 {
722 1 : if vs.manifestFileNum != 0 {
723 1 : vs.obsoleteManifests = append(vs.obsoleteManifests, obsoleteFile{
724 1 : fileType: base.FileTypeManifest,
725 1 : fs: vs.fs,
726 1 : path: base.MakeFilepath(vs.fs, vs.dirname, base.FileTypeManifest, vs.manifestFileNum),
727 1 : fileNum: vs.manifestFileNum,
728 1 : fileSize: prevManifestFileSize,
729 1 : isLocal: true,
730 1 : })
731 1 : }
732 1 : vs.manifestFileNum = newManifestFileNum
733 : }
734 :
735 1 : vs.metrics.updateLevelMetrics(vu.Metrics)
736 1 : for i := range vs.metrics.Levels {
737 1 : l := &vs.metrics.Levels[i]
738 1 : l.TablesCount = int64(newVersion.Levels[i].Len())
739 1 : l.VirtualTablesCount = newVersion.Levels[i].NumVirtual
740 1 : l.VirtualTablesSize = newVersion.Levels[i].VirtualTableSize
741 1 : l.TablesSize = int64(newVersion.Levels[i].TableSize())
742 1 : l.EstimatedReferencesSize = newVersion.Levels[i].EstimatedReferenceSize()
743 1 : l.Sublevels = 0
744 1 : if l.TablesCount > 0 {
745 1 : l.Sublevels = 1
746 1 : }
747 1 : if invariants.Enabled {
748 1 : levelFiles := newVersion.Levels[i].Slice()
749 1 : if size := int64(levelFiles.TableSizeSum()); l.TablesSize != size {
750 0 : vs.opts.Logger.Fatalf("versionSet metrics L%d Size = %d, actual size = %d", i, l.TablesSize, size)
751 0 : }
752 1 : refSize := uint64(0)
753 1 : for f := range levelFiles.All() {
754 1 : refSize += f.EstimatedReferenceSize()
755 1 : }
756 1 : if refSize != l.EstimatedReferencesSize {
757 0 : vs.opts.Logger.Fatalf("versionSet metrics L%d EstimatedReferencesSize = %d, recomputed size = %d", i, l.EstimatedReferencesSize, refSize)
758 0 : }
759 :
760 1 : if nVirtual := levelFiles.NumVirtual(); nVirtual != l.VirtualTablesCount {
761 0 : vs.opts.Logger.Fatalf(
762 0 : "versionSet metrics L%d NumVirtual = %d, actual NumVirtual = %d",
763 0 : i, l.VirtualTablesCount, nVirtual,
764 0 : )
765 0 : }
766 1 : if vSize := levelFiles.VirtualTableSizeSum(); vSize != l.VirtualTablesSize {
767 0 : vs.opts.Logger.Fatalf(
768 0 : "versionSet metrics L%d Virtual size = %d, actual size = %d",
769 0 : i, l.VirtualTablesSize, vSize,
770 0 : )
771 0 : }
772 : }
773 : }
774 1 : vs.metrics.Levels[0].Sublevels = int32(len(newVersion.L0SublevelFiles))
775 1 : vs.metrics.Table.Local.LiveSize = uint64(int64(vs.metrics.Table.Local.LiveSize) + localTablesLiveDelta.size)
776 1 : vs.metrics.Table.Local.LiveCount = uint64(int64(vs.metrics.Table.Local.LiveCount) + localTablesLiveDelta.count)
777 1 : vs.metrics.BlobFiles.Local.LiveSize = uint64(int64(vs.metrics.BlobFiles.Local.LiveSize) + localBlobLiveDelta.size)
778 1 : vs.metrics.BlobFiles.Local.LiveCount = uint64(int64(vs.metrics.BlobFiles.Local.LiveCount) + localBlobLiveDelta.count)
779 1 :
780 1 : vs.setCompactionPicker(
781 1 : newCompactionPickerByScore(newVersion, vs.l0Organizer, &vs.virtualBackings, vs.opts, inProgress))
782 1 : if !vs.dynamicBaseLevel {
783 1 : vs.picker.forceBaseLevel1()
784 1 : }
785 1 : return nil
786 : }
787 :
788 1 : func (vs *versionSet) setCompactionPicker(picker *compactionPickerByScore) {
789 1 : vs.picker = picker
790 1 : vs.curCompactionConcurrency.Store(int32(picker.getCompactionConcurrency()))
791 1 : }
792 :
793 : type fileBackingInfo struct {
794 : backing *fileBacking
795 : isLocal bool
796 : }
797 :
798 : type fileMetricDelta struct {
799 : count int64
800 : size int64
801 : }
802 :
803 : // getZombieTablesAndUpdateVirtualBackings updates the virtual backings with the
804 : // changes in the versionEdit and populates ve.RemovedBackingTables.
805 : // Returns:
806 : // - zombieBackings: all backings (physical and virtual) that will no longer be
807 : // needed when we apply ve.
808 : // - removedVirtualBackings: the virtual backings that will be removed by the
809 : // VersionEdit and which must be Unref()ed by the caller. These backings
810 : // match ve.RemovedBackingTables.
811 : // - localLiveSizeDelta: the delta in local live bytes.
812 : func getZombieTablesAndUpdateVirtualBackings(
813 : ve *versionEdit, virtualBackings *manifest.VirtualBackings, provider objstorage.Provider,
814 1 : ) (zombieBackings, removedVirtualBackings []fileBackingInfo, localLiveDelta fileMetricDelta) {
815 1 : // First, deal with the physical tables.
816 1 : //
817 1 : // A physical backing has become unused if it is in DeletedFiles but not in
818 1 : // NewFiles or CreatedBackingTables.
819 1 : //
820 1 : // Note that for the common case where there are very few elements, the map
821 1 : // will stay on the stack.
822 1 : stillUsed := make(map[base.DiskFileNum]struct{})
823 1 : for _, nf := range ve.NewTables {
824 1 : if !nf.Meta.Virtual {
825 1 : stillUsed[nf.Meta.FileBacking.DiskFileNum] = struct{}{}
826 1 : isLocal, localFileDelta := sizeIfLocal(nf.Meta.FileBacking, provider)
827 1 : localLiveDelta.size += localFileDelta
828 1 : if isLocal {
829 1 : localLiveDelta.count++
830 1 : }
831 : }
832 : }
833 1 : for _, b := range ve.CreatedBackingTables {
834 1 : stillUsed[b.DiskFileNum] = struct{}{}
835 1 : }
836 1 : for _, m := range ve.DeletedTables {
837 1 : if !m.Virtual {
838 1 : // NB: this deleted file may also be in NewFiles or
839 1 : // CreatedBackingTables, due to a file moving between levels, or
840 1 : // becoming virtualized. In which case there is no change due to this
841 1 : // file in the localLiveSizeDelta -- the subtraction below compensates
842 1 : // for the addition.
843 1 : isLocal, localFileDelta := sizeIfLocal(m.FileBacking, provider)
844 1 : localLiveDelta.size -= localFileDelta
845 1 : if isLocal {
846 1 : localLiveDelta.count--
847 1 : }
848 1 : if _, ok := stillUsed[m.FileBacking.DiskFileNum]; !ok {
849 1 : zombieBackings = append(zombieBackings, fileBackingInfo{
850 1 : backing: m.FileBacking,
851 1 : isLocal: isLocal,
852 1 : })
853 1 : }
854 : }
855 : }
856 :
857 : // Now deal with virtual tables.
858 : //
859 : // When a virtual table moves between levels we AddTable() then RemoveTable(),
860 : // which works out.
861 1 : for _, b := range ve.CreatedBackingTables {
862 1 : virtualBackings.AddAndRef(b)
863 1 : isLocal, localFileDelta := sizeIfLocal(b, provider)
864 1 : localLiveDelta.size += localFileDelta
865 1 : if isLocal {
866 1 : localLiveDelta.count++
867 1 : }
868 : }
869 1 : for _, nf := range ve.NewTables {
870 1 : if nf.Meta.Virtual {
871 1 : virtualBackings.AddTable(nf.Meta)
872 1 : }
873 : }
874 1 : for _, m := range ve.DeletedTables {
875 1 : if m.Virtual {
876 1 : virtualBackings.RemoveTable(m)
877 1 : }
878 : }
879 :
880 1 : if unused := virtualBackings.Unused(); len(unused) > 0 {
881 1 : // Virtual backings that are no longer used are zombies and are also added
882 1 : // to RemovedBackingTables (before the version edit is written to disk).
883 1 : ve.RemovedBackingTables = make([]base.DiskFileNum, len(unused))
884 1 : for i, b := range unused {
885 1 : isLocal, localFileDelta := sizeIfLocal(b, provider)
886 1 : localLiveDelta.size -= localFileDelta
887 1 : if isLocal {
888 1 : localLiveDelta.count--
889 1 : }
890 1 : ve.RemovedBackingTables[i] = b.DiskFileNum
891 1 : zombieBackings = append(zombieBackings, fileBackingInfo{
892 1 : backing: b,
893 1 : isLocal: isLocal,
894 1 : })
895 1 : virtualBackings.Remove(b.DiskFileNum)
896 : }
897 1 : removedVirtualBackings = zombieBackings[len(zombieBackings)-len(unused):]
898 : }
899 1 : return zombieBackings, removedVirtualBackings, localLiveDelta
900 : }
901 :
902 : // getZombieBlobFilesAndComputeLocalMetrics constructs objectInfos for all
903 : // zombie blob files, and computes the metric deltas for live files overall and
904 : // locally.
905 : func getZombieBlobFilesAndComputeLocalMetrics(
906 : ve *versionEdit, provider objstorage.Provider,
907 1 : ) (zombieBlobFiles []objectInfo, localLiveDelta fileMetricDelta) {
908 1 : for _, b := range ve.NewBlobFiles {
909 1 : if objstorage.IsLocalBlobFile(provider, b.FileNum) {
910 1 : localLiveDelta.count++
911 1 : localLiveDelta.size += int64(b.Size)
912 1 : }
913 : }
914 1 : zombieBlobFiles = make([]objectInfo, 0, len(ve.DeletedBlobFiles))
915 1 : for _, b := range ve.DeletedBlobFiles {
916 1 : isLocal := objstorage.IsLocalBlobFile(provider, b.FileNum)
917 1 : if isLocal {
918 1 : localLiveDelta.count--
919 1 : localLiveDelta.size -= int64(b.Size)
920 1 : }
921 1 : zombieBlobFiles = append(zombieBlobFiles, objectInfo{
922 1 : fileInfo: fileInfo{
923 1 : FileNum: b.FileNum,
924 1 : FileSize: b.Size,
925 1 : },
926 1 : isLocal: isLocal,
927 1 : })
928 : }
929 1 : return zombieBlobFiles, localLiveDelta
930 : }
931 :
932 : // sizeIfLocal returns backing.Size if the backing is a local file, else 0.
933 : func sizeIfLocal(
934 : backing *fileBacking, provider objstorage.Provider,
935 1 : ) (isLocal bool, localSize int64) {
936 1 : isLocal = objstorage.IsLocalTable(provider, backing.DiskFileNum)
937 1 : if isLocal {
938 1 : return true, int64(backing.Size)
939 1 : }
940 1 : return false, 0
941 : }
942 :
943 : func (vs *versionSet) incrementCompactions(
944 : kind compactionKind,
945 : extraLevels []*compactionLevel,
946 : pickerMetrics pickedCompactionMetrics,
947 : bytesWritten int64,
948 : compactionErr error,
949 1 : ) {
950 1 : if kind == compactionKindFlush || kind == compactionKindIngestedFlushable {
951 1 : vs.metrics.Flush.Count++
952 1 : } else {
953 1 : vs.metrics.Compact.Count++
954 1 : if compactionErr != nil {
955 1 : if errors.Is(compactionErr, ErrCancelledCompaction) {
956 1 : vs.metrics.Compact.CancelledCount++
957 1 : vs.metrics.Compact.CancelledBytes += bytesWritten
958 1 : } else {
959 1 : vs.metrics.Compact.FailedCount++
960 1 : }
961 : }
962 : }
963 :
964 1 : switch kind {
965 1 : case compactionKindDefault:
966 1 : vs.metrics.Compact.DefaultCount++
967 :
968 1 : case compactionKindFlush, compactionKindIngestedFlushable:
969 :
970 1 : case compactionKindMove:
971 1 : vs.metrics.Compact.MoveCount++
972 :
973 1 : case compactionKindDeleteOnly:
974 1 : vs.metrics.Compact.DeleteOnlyCount++
975 :
976 1 : case compactionKindElisionOnly:
977 1 : vs.metrics.Compact.ElisionOnlyCount++
978 :
979 1 : case compactionKindRead:
980 1 : vs.metrics.Compact.ReadCount++
981 :
982 1 : case compactionKindTombstoneDensity:
983 1 : vs.metrics.Compact.TombstoneDensityCount++
984 :
985 1 : case compactionKindRewrite:
986 1 : vs.metrics.Compact.RewriteCount++
987 :
988 1 : case compactionKindCopy:
989 1 : vs.metrics.Compact.CopyCount++
990 :
991 0 : default:
992 0 : if invariants.Enabled {
993 0 : panic("unhandled compaction kind")
994 : }
995 : }
996 1 : if len(extraLevels) > 0 {
997 1 : vs.metrics.Compact.MultiLevelCount++
998 1 : }
999 : }
1000 :
1001 1 : func (vs *versionSet) incrementCompactionBytes(numBytes int64) {
1002 1 : vs.atomicInProgressBytes.Add(numBytes)
1003 1 : }
1004 :
1005 : // createManifest creates a manifest file that contains a snapshot of vs.
1006 : func (vs *versionSet) createManifest(
1007 : dirname string,
1008 : fileNum, minUnflushedLogNum base.DiskFileNum,
1009 : nextFileNum uint64,
1010 : virtualBackings []*fileBacking,
1011 1 : ) (err error) {
1012 1 : var (
1013 1 : filename = base.MakeFilepath(vs.fs, dirname, base.FileTypeManifest, fileNum)
1014 1 : manifestFile vfs.File
1015 1 : manifestWriter *record.Writer
1016 1 : )
1017 1 : defer func() {
1018 1 : if manifestWriter != nil {
1019 0 : _ = manifestWriter.Close()
1020 0 : }
1021 1 : if manifestFile != nil {
1022 0 : _ = manifestFile.Close()
1023 0 : }
1024 1 : if err != nil {
1025 1 : _ = vs.fs.Remove(filename)
1026 1 : }
1027 : }()
1028 1 : manifestFile, err = vs.fs.Create(filename, "pebble-manifest")
1029 1 : if err != nil {
1030 1 : return err
1031 1 : }
1032 1 : manifestWriter = record.NewWriter(manifestFile)
1033 1 :
1034 1 : snapshot := manifest.VersionEdit{
1035 1 : ComparerName: vs.cmp.Name,
1036 1 : // When creating a version snapshot for an existing DB, this snapshot
1037 1 : // VersionEdit will be immediately followed by another VersionEdit
1038 1 : // (being written in UpdateVersionLocked()). That VersionEdit always
1039 1 : // contains a LastSeqNum, so we don't need to include that in the
1040 1 : // snapshot. But it does not necessarily include MinUnflushedLogNum,
1041 1 : // NextFileNum, so we initialize those using the corresponding fields in
1042 1 : // the versionSet (which came from the latest preceding VersionEdit that
1043 1 : // had those fields).
1044 1 : MinUnflushedLogNum: minUnflushedLogNum,
1045 1 : NextFileNum: nextFileNum,
1046 1 : CreatedBackingTables: virtualBackings,
1047 1 : NewBlobFiles: vs.blobFiles.Metadatas(),
1048 1 : }
1049 1 : // Add all extant sstables in the current version.
1050 1 : for level, levelMetadata := range vs.currentVersion().Levels {
1051 1 : for meta := range levelMetadata.All() {
1052 1 : snapshot.NewTables = append(snapshot.NewTables, newTableEntry{
1053 1 : Level: level,
1054 1 : Meta: meta,
1055 1 : })
1056 1 : }
1057 : }
1058 :
1059 1 : w, err1 := manifestWriter.Next()
1060 1 : if err1 != nil {
1061 0 : return err1
1062 0 : }
1063 1 : if err := snapshot.Encode(w); err != nil {
1064 0 : return err
1065 0 : }
1066 :
1067 1 : if vs.manifest != nil {
1068 1 : if err := vs.manifest.Close(); err != nil {
1069 0 : return err
1070 0 : }
1071 1 : vs.manifest = nil
1072 : }
1073 1 : if vs.manifestFile != nil {
1074 1 : if err := vs.manifestFile.Close(); err != nil {
1075 0 : return err
1076 0 : }
1077 1 : vs.manifestFile = nil
1078 : }
1079 :
1080 1 : vs.manifest, manifestWriter = manifestWriter, nil
1081 1 : vs.manifestFile, manifestFile = manifestFile, nil
1082 1 : return nil
1083 : }
1084 :
1085 : // NB: This method is not safe for concurrent use. It is only safe
1086 : // to be called when concurrent changes to nextFileNum are not expected.
1087 1 : func (vs *versionSet) markFileNumUsed(fileNum base.DiskFileNum) {
1088 1 : if vs.nextFileNum.Load() <= uint64(fileNum) {
1089 1 : vs.nextFileNum.Store(uint64(fileNum + 1))
1090 1 : }
1091 : }
1092 :
1093 : // getNextFileNum returns the next file number to be used.
1094 : //
1095 : // Can be called without the versionSet's mutex being held.
1096 1 : func (vs *versionSet) getNextFileNum() base.FileNum {
1097 1 : x := vs.nextFileNum.Add(1) - 1
1098 1 : return base.FileNum(x)
1099 1 : }
1100 :
1101 : // Can be called without the versionSet's mutex being held.
1102 1 : func (vs *versionSet) getNextDiskFileNum() base.DiskFileNum {
1103 1 : x := vs.nextFileNum.Add(1) - 1
1104 1 : return base.DiskFileNum(x)
1105 1 : }
1106 :
1107 1 : func (vs *versionSet) append(v *version) {
1108 1 : if v.Refs() != 0 {
1109 0 : panic("pebble: version should be unreferenced")
1110 : }
1111 1 : if !vs.versions.Empty() {
1112 1 : vs.versions.Back().UnrefLocked()
1113 1 : }
1114 1 : v.Deleted = vs.obsoleteFn
1115 1 : v.Ref()
1116 1 : vs.versions.PushBack(v)
1117 1 : if invariants.Enabled {
1118 1 : // Verify that the virtualBackings contains all the backings referenced by
1119 1 : // the version.
1120 1 : for _, l := range v.Levels {
1121 1 : for f := range l.All() {
1122 1 : if f.Virtual {
1123 1 : if _, ok := vs.virtualBackings.Get(f.FileBacking.DiskFileNum); !ok {
1124 0 : panic(fmt.Sprintf("%s is not in virtualBackings", f.FileBacking.DiskFileNum))
1125 : }
1126 : }
1127 : }
1128 : }
1129 : }
1130 : }
1131 :
1132 1 : func (vs *versionSet) currentVersion() *version {
1133 1 : return vs.versions.Back()
1134 1 : }
1135 :
1136 1 : func (vs *versionSet) addLiveFileNums(m map[base.DiskFileNum]struct{}) {
1137 1 : current := vs.currentVersion()
1138 1 : for v := vs.versions.Front(); true; v = v.Next() {
1139 1 : for _, lm := range v.Levels {
1140 1 : for f := range lm.All() {
1141 1 : m[f.FileBacking.DiskFileNum] = struct{}{}
1142 1 : for _, ref := range f.BlobReferences {
1143 0 : m[ref.FileNum] = struct{}{}
1144 0 : }
1145 : }
1146 : }
1147 1 : if v == current {
1148 1 : break
1149 : }
1150 : }
1151 : // virtualBackings contains backings that are referenced by some virtual
1152 : // tables in the latest version (which are handled above), and backings that
1153 : // are not but are still alive because of the protection mechanism (see
1154 : // manifset.VirtualBackings). This loop ensures the latter get added to the
1155 : // map.
1156 1 : vs.virtualBackings.ForEach(func(b *fileBacking) {
1157 1 : m[b.DiskFileNum] = struct{}{}
1158 1 : })
1159 : }
1160 :
1161 : // addObsoleteLocked will add the fileInfo associated with obsolete backing
1162 : // sstables to the obsolete tables list.
1163 : //
1164 : // The file backings in the obsolete list must not appear more than once.
1165 : //
1166 : // DB.mu must be held when addObsoleteLocked is called.
1167 1 : func (vs *versionSet) addObsoleteLocked(obsolete manifest.ObsoleteFiles) {
1168 1 : if obsolete.Count() == 0 {
1169 1 : return
1170 1 : }
1171 :
1172 : // Note that the zombie objects transition from zombie *to* obsolete, and
1173 : // will no longer be considered zombie.
1174 :
1175 1 : newlyObsoleteTables := make([]objectInfo, len(obsolete.FileBackings))
1176 1 : for i, bs := range obsolete.FileBackings {
1177 1 : newlyObsoleteTables[i] = vs.zombieTables.Extract(bs.DiskFileNum)
1178 1 : }
1179 1 : vs.obsoleteTables = mergeObjectInfos(vs.obsoleteTables, newlyObsoleteTables)
1180 1 :
1181 1 : newlyObsoleteBlobFiles := make([]objectInfo, len(obsolete.BlobFiles))
1182 1 : for i, bf := range obsolete.BlobFiles {
1183 1 : newlyObsoleteBlobFiles[i] = vs.zombieBlobs.Extract(bf.FileNum)
1184 1 : }
1185 1 : vs.obsoleteBlobs = mergeObjectInfos(vs.obsoleteBlobs, newlyObsoleteBlobFiles)
1186 1 : vs.updateObsoleteObjectMetricsLocked()
1187 : }
1188 :
1189 : // addObsolete will acquire DB.mu, so DB.mu must not be held when this is
1190 : // called.
1191 1 : func (vs *versionSet) addObsolete(obsolete manifest.ObsoleteFiles) {
1192 1 : vs.mu.Lock()
1193 1 : defer vs.mu.Unlock()
1194 1 : vs.addObsoleteLocked(obsolete)
1195 1 : }
1196 :
1197 1 : func (vs *versionSet) updateObsoleteObjectMetricsLocked() {
1198 1 : vs.metrics.Table.ObsoleteCount = int64(len(vs.obsoleteTables))
1199 1 : vs.metrics.Table.ObsoleteSize = 0
1200 1 : vs.metrics.Table.Local.ObsoleteSize = 0
1201 1 : vs.metrics.Table.Local.ObsoleteCount = 0
1202 1 : for _, fi := range vs.obsoleteTables {
1203 1 : vs.metrics.Table.ObsoleteSize += fi.FileSize
1204 1 : if fi.isLocal {
1205 1 : vs.metrics.Table.Local.ObsoleteSize += fi.FileSize
1206 1 : vs.metrics.Table.Local.ObsoleteCount++
1207 1 : }
1208 : }
1209 1 : vs.metrics.BlobFiles.ObsoleteCount = uint64(len(vs.obsoleteBlobs))
1210 1 : vs.metrics.BlobFiles.ObsoleteSize = 0
1211 1 : vs.metrics.BlobFiles.Local.ObsoleteSize = 0
1212 1 : vs.metrics.BlobFiles.Local.ObsoleteCount = 0
1213 1 : for _, fi := range vs.obsoleteBlobs {
1214 1 : vs.metrics.BlobFiles.ObsoleteSize += fi.FileSize
1215 1 : if fi.isLocal {
1216 1 : vs.metrics.BlobFiles.Local.ObsoleteSize += fi.FileSize
1217 1 : vs.metrics.BlobFiles.Local.ObsoleteCount++
1218 1 : }
1219 : }
1220 : }
1221 :
1222 : func findCurrentManifest(
1223 : fs vfs.FS, dirname string, ls []string,
1224 1 : ) (marker *atomicfs.Marker, manifestNum base.DiskFileNum, exists bool, err error) {
1225 1 : // Locating a marker should succeed even if the marker has never been placed.
1226 1 : var filename string
1227 1 : marker, filename, err = atomicfs.LocateMarkerInListing(fs, dirname, manifestMarkerName, ls)
1228 1 : if err != nil {
1229 1 : return nil, 0, false, err
1230 1 : }
1231 :
1232 1 : if filename == "" {
1233 1 : // The marker hasn't been set yet. This database doesn't exist.
1234 1 : return marker, 0, false, nil
1235 1 : }
1236 :
1237 1 : var ok bool
1238 1 : _, manifestNum, ok = base.ParseFilename(fs, filename)
1239 1 : if !ok {
1240 0 : return marker, 0, false, base.CorruptionErrorf("pebble: MANIFEST name %q is malformed", errors.Safe(filename))
1241 0 : }
1242 1 : return marker, manifestNum, true, nil
1243 : }
1244 :
1245 1 : func newFileMetrics(newFiles []manifest.NewTableEntry) levelMetricsDelta {
1246 1 : var m levelMetricsDelta
1247 1 : for _, nf := range newFiles {
1248 1 : lm := m[nf.Level]
1249 1 : if lm == nil {
1250 1 : lm = &LevelMetrics{}
1251 1 : m[nf.Level] = lm
1252 1 : }
1253 1 : lm.TablesCount++
1254 1 : lm.TablesSize += int64(nf.Meta.Size)
1255 1 : lm.EstimatedReferencesSize += nf.Meta.EstimatedReferenceSize()
1256 : }
1257 1 : return m
1258 : }
|
