Line data Source code
1 : // Copyright 2018 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 : "context"
9 : "slices"
10 : "sort"
11 : "time"
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/keyspan"
17 : "github.com/cockroachdb/pebble/internal/manifest"
18 : "github.com/cockroachdb/pebble/internal/private"
19 : "github.com/cockroachdb/pebble/objstorage"
20 : "github.com/cockroachdb/pebble/objstorage/remote"
21 : "github.com/cockroachdb/pebble/sstable"
22 : )
23 :
24 2 : func sstableKeyCompare(userCmp Compare, a, b InternalKey) int {
25 2 : c := userCmp(a.UserKey, b.UserKey)
26 2 : if c != 0 {
27 2 : return c
28 2 : }
29 2 : if a.IsExclusiveSentinel() {
30 2 : if !b.IsExclusiveSentinel() {
31 2 : return -1
32 2 : }
33 2 : } else if b.IsExclusiveSentinel() {
34 2 : return +1
35 2 : }
36 2 : return 0
37 : }
38 :
39 : // KeyRange encodes a key range in user key space. A KeyRange's Start is
40 : // inclusive while its End is exclusive.
41 : type KeyRange struct {
42 : Start, End []byte
43 : }
44 :
45 : // Valid returns true if the KeyRange is defined.
46 2 : func (k *KeyRange) Valid() bool {
47 2 : return k.Start != nil && k.End != nil
48 2 : }
49 :
50 : // Contains returns whether the specified key exists in the KeyRange.
51 2 : func (k *KeyRange) Contains(cmp base.Compare, key InternalKey) bool {
52 2 : v := cmp(key.UserKey, k.End)
53 2 : return (v < 0 || (v == 0 && key.IsExclusiveSentinel())) && cmp(k.Start, key.UserKey) <= 0
54 2 : }
55 :
56 : // OverlapsInternalKeyRange checks if the specified internal key range has an
57 : // overlap with the KeyRange. Note that we aren't checking for full containment
58 : // of smallest-largest within k, rather just that there's some intersection
59 : // between the two ranges.
60 2 : func (k *KeyRange) OverlapsInternalKeyRange(cmp base.Compare, smallest, largest InternalKey) bool {
61 2 : v := cmp(k.Start, largest.UserKey)
62 2 : return v <= 0 && !(largest.IsExclusiveSentinel() && v == 0) &&
63 2 : cmp(k.End, smallest.UserKey) > 0
64 2 : }
65 :
66 : // Overlaps checks if the specified file has an overlap with the KeyRange.
67 : // Note that we aren't checking for full containment of m within k, rather just
68 : // that there's some intersection between m and k's bounds.
69 2 : func (k *KeyRange) Overlaps(cmp base.Compare, m *fileMetadata) bool {
70 2 : return k.OverlapsInternalKeyRange(cmp, m.Smallest, m.Largest)
71 2 : }
72 :
73 : // OverlapsKeyRange checks if this span overlaps with the provided KeyRange.
74 : // Note that we aren't checking for full containment of either span in the other,
75 : // just that there's a key x that is in both key ranges.
76 2 : func (k *KeyRange) OverlapsKeyRange(cmp Compare, span KeyRange) bool {
77 2 : return cmp(k.Start, span.End) < 0 && cmp(k.End, span.Start) > 0
78 2 : }
79 :
80 2 : func ingestValidateKey(opts *Options, key *InternalKey) error {
81 2 : if key.Kind() == InternalKeyKindInvalid {
82 1 : return base.CorruptionErrorf("pebble: external sstable has corrupted key: %s",
83 1 : key.Pretty(opts.Comparer.FormatKey))
84 1 : }
85 2 : if key.SeqNum() != 0 {
86 1 : return base.CorruptionErrorf("pebble: external sstable has non-zero seqnum: %s",
87 1 : key.Pretty(opts.Comparer.FormatKey))
88 1 : }
89 2 : return nil
90 : }
91 :
92 : // ingestSynthesizeShared constructs a fileMetadata for one shared sstable owned
93 : // or shared by another node.
94 : func ingestSynthesizeShared(
95 : opts *Options, sm SharedSSTMeta, fileNum base.DiskFileNum,
96 2 : ) (*fileMetadata, error) {
97 2 : if sm.Size == 0 {
98 0 : // Disallow 0 file sizes
99 0 : return nil, errors.New("pebble: cannot ingest shared file with size 0")
100 0 : }
101 : // Don't load table stats. Doing a round trip to shared storage, one SST
102 : // at a time is not worth it as it slows down ingestion.
103 2 : meta := &fileMetadata{
104 2 : FileNum: fileNum.FileNum(),
105 2 : CreationTime: time.Now().Unix(),
106 2 : Virtual: true,
107 2 : Size: sm.Size,
108 2 : }
109 2 : meta.InitProviderBacking(fileNum)
110 2 : // Set the underlying FileBacking's size to the same size as the virtualized
111 2 : // view of the sstable. This ensures that we don't over-prioritize this
112 2 : // sstable for compaction just yet, as we do not have a clear sense of what
113 2 : // parts of this sstable are referenced by other nodes.
114 2 : meta.FileBacking.Size = sm.Size
115 2 : if sm.LargestRangeKey.Valid() && sm.LargestRangeKey.UserKey != nil {
116 2 : // Initialize meta.{HasRangeKeys,Smallest,Largest}, etc.
117 2 : //
118 2 : // NB: We create new internal keys and pass them into ExternalRangeKeyBounds
119 2 : // so that we can sub a zero sequence number into the bounds. We can set
120 2 : // the sequence number to anything here; it'll be reset in ingestUpdateSeqNum
121 2 : // anyway. However we do need to use the same sequence number across all
122 2 : // bound keys at this step so that we end up with bounds that are consistent
123 2 : // across point/range keys.
124 2 : smallestRangeKey := base.MakeInternalKey(sm.SmallestRangeKey.UserKey, 0, sm.SmallestRangeKey.Kind())
125 2 : largestRangeKey := base.MakeExclusiveSentinelKey(sm.LargestRangeKey.Kind(), sm.LargestRangeKey.UserKey)
126 2 : meta.ExtendRangeKeyBounds(opts.Comparer.Compare, smallestRangeKey, largestRangeKey)
127 2 : }
128 2 : if sm.LargestPointKey.Valid() && sm.LargestPointKey.UserKey != nil {
129 2 : // Initialize meta.{HasPointKeys,Smallest,Largest}, etc.
130 2 : //
131 2 : // See point above in the ExtendRangeKeyBounds call on why we use a zero
132 2 : // sequence number here.
133 2 : smallestPointKey := base.MakeInternalKey(sm.SmallestPointKey.UserKey, 0, sm.SmallestPointKey.Kind())
134 2 : largestPointKey := base.MakeInternalKey(sm.LargestPointKey.UserKey, 0, sm.LargestPointKey.Kind())
135 2 : if sm.LargestPointKey.IsExclusiveSentinel() {
136 2 : largestPointKey = base.MakeRangeDeleteSentinelKey(sm.LargestPointKey.UserKey)
137 2 : }
138 2 : meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallestPointKey, largestPointKey)
139 : }
140 2 : if err := meta.Validate(opts.Comparer.Compare, opts.Comparer.FormatKey); err != nil {
141 0 : return nil, err
142 0 : }
143 2 : return meta, nil
144 : }
145 :
146 : // ingestLoad1External loads the fileMetadata for one external sstable.
147 : // Sequence number and target level calculation happens during prepare/apply.
148 : func ingestLoad1External(
149 : opts *Options,
150 : e ExternalFile,
151 : fileNum base.DiskFileNum,
152 : objprovider objstorage.Provider,
153 : jobID int,
154 1 : ) (*fileMetadata, error) {
155 1 : if e.Size == 0 {
156 0 : // Disallow 0 file sizes
157 0 : return nil, errors.New("pebble: cannot ingest external file with size 0")
158 0 : }
159 1 : if !e.HasRangeKey && !e.HasPointKey {
160 0 : return nil, errors.New("pebble: cannot ingest external file with no point or range keys")
161 0 : }
162 : // Don't load table stats. Doing a round trip to shared storage, one SST
163 : // at a time is not worth it as it slows down ingestion.
164 1 : meta := &fileMetadata{}
165 1 : meta.FileNum = fileNum.FileNum()
166 1 : meta.CreationTime = time.Now().Unix()
167 1 : meta.Virtual = true
168 1 : meta.Size = e.Size
169 1 : meta.InitProviderBacking(fileNum)
170 1 :
171 1 : // Try to resolve a reference to the external file.
172 1 : backing, err := objprovider.CreateExternalObjectBacking(e.Locator, e.ObjName)
173 1 : if err != nil {
174 0 : return nil, err
175 0 : }
176 1 : metas, err := objprovider.AttachRemoteObjects([]objstorage.RemoteObjectToAttach{{
177 1 : FileNum: fileNum,
178 1 : FileType: fileTypeTable,
179 1 : Backing: backing,
180 1 : }})
181 1 : if err != nil {
182 0 : return nil, err
183 0 : }
184 1 : if opts.EventListener.TableCreated != nil {
185 1 : opts.EventListener.TableCreated(TableCreateInfo{
186 1 : JobID: jobID,
187 1 : Reason: "ingesting",
188 1 : Path: objprovider.Path(metas[0]),
189 1 : FileNum: fileNum.FileNum(),
190 1 : })
191 1 : }
192 : // In the name of keeping this ingestion as fast as possible, we avoid
193 : // *all* existence checks and synthesize a file metadata with smallest/largest
194 : // keys that overlap whatever the passed-in span was.
195 1 : smallestCopy := make([]byte, len(e.SmallestUserKey))
196 1 : copy(smallestCopy, e.SmallestUserKey)
197 1 : largestCopy := make([]byte, len(e.LargestUserKey))
198 1 : copy(largestCopy, e.LargestUserKey)
199 1 : if e.HasPointKey {
200 1 : meta.ExtendPointKeyBounds(opts.Comparer.Compare, base.MakeInternalKey(smallestCopy, 0, InternalKeyKindMax),
201 1 : base.MakeRangeDeleteSentinelKey(largestCopy))
202 1 : }
203 1 : if e.HasRangeKey {
204 0 : meta.ExtendRangeKeyBounds(opts.Comparer.Compare, base.MakeInternalKey(smallestCopy, 0, InternalKeyKindRangeKeySet),
205 0 : base.MakeExclusiveSentinelKey(InternalKeyKindRangeKeyDelete, largestCopy))
206 0 : }
207 :
208 : // Set the underlying FileBacking's size to the same size as the virtualized
209 : // view of the sstable. This ensures that we don't over-prioritize this
210 : // sstable for compaction just yet, as we do not have a clear sense of
211 : // what parts of this sstable are referenced by other nodes.
212 1 : meta.FileBacking.Size = e.Size
213 1 :
214 1 : if err := meta.Validate(opts.Comparer.Compare, opts.Comparer.FormatKey); err != nil {
215 0 : return nil, err
216 0 : }
217 1 : return meta, nil
218 : }
219 :
220 : // ingestLoad1 creates the FileMetadata for one file. This file will be owned
221 : // by this store.
222 : func ingestLoad1(
223 : opts *Options,
224 : fmv FormatMajorVersion,
225 : readable objstorage.Readable,
226 : cacheID uint64,
227 : fileNum base.DiskFileNum,
228 2 : ) (*fileMetadata, error) {
229 2 : cacheOpts := private.SSTableCacheOpts(cacheID, fileNum).(sstable.ReaderOption)
230 2 : r, err := sstable.NewReader(readable, opts.MakeReaderOptions(), cacheOpts)
231 2 : if err != nil {
232 1 : return nil, err
233 1 : }
234 2 : defer r.Close()
235 2 :
236 2 : // Avoid ingesting tables with format versions this DB doesn't support.
237 2 : tf, err := r.TableFormat()
238 2 : if err != nil {
239 0 : return nil, err
240 0 : }
241 2 : if tf < fmv.MinTableFormat() || tf > fmv.MaxTableFormat() {
242 1 : return nil, errors.Newf(
243 1 : "pebble: table format %s is not within range supported at DB format major version %d, (%s,%s)",
244 1 : tf, fmv, fmv.MinTableFormat(), fmv.MaxTableFormat(),
245 1 : )
246 1 : }
247 :
248 2 : meta := &fileMetadata{}
249 2 : meta.FileNum = fileNum.FileNum()
250 2 : meta.Size = uint64(readable.Size())
251 2 : meta.CreationTime = time.Now().Unix()
252 2 : meta.InitPhysicalBacking()
253 2 :
254 2 : // Avoid loading into the table cache for collecting stats if we
255 2 : // don't need to. If there are no range deletions, we have all the
256 2 : // information to compute the stats here.
257 2 : //
258 2 : // This is helpful in tests for avoiding awkwardness around deletion of
259 2 : // ingested files from MemFS. MemFS implements the Windows semantics of
260 2 : // disallowing removal of an open file. Under MemFS, if we don't populate
261 2 : // meta.Stats here, the file will be loaded into the table cache for
262 2 : // calculating stats before we can remove the original link.
263 2 : maybeSetStatsFromProperties(meta.PhysicalMeta(), &r.Properties)
264 2 :
265 2 : {
266 2 : iter, err := r.NewIter(nil /* lower */, nil /* upper */)
267 2 : if err != nil {
268 1 : return nil, err
269 1 : }
270 2 : defer iter.Close()
271 2 : var smallest InternalKey
272 2 : if key, _ := iter.First(); key != nil {
273 2 : if err := ingestValidateKey(opts, key); err != nil {
274 1 : return nil, err
275 1 : }
276 2 : smallest = (*key).Clone()
277 : }
278 2 : if err := iter.Error(); err != nil {
279 1 : return nil, err
280 1 : }
281 2 : if key, _ := iter.Last(); key != nil {
282 2 : if err := ingestValidateKey(opts, key); err != nil {
283 0 : return nil, err
284 0 : }
285 2 : meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallest, key.Clone())
286 : }
287 2 : if err := iter.Error(); err != nil {
288 1 : return nil, err
289 1 : }
290 : }
291 :
292 2 : iter, err := r.NewRawRangeDelIter()
293 2 : if err != nil {
294 0 : return nil, err
295 0 : }
296 2 : if iter != nil {
297 2 : defer iter.Close()
298 2 : var smallest InternalKey
299 2 : if s := iter.First(); s != nil {
300 2 : key := s.SmallestKey()
301 2 : if err := ingestValidateKey(opts, &key); err != nil {
302 0 : return nil, err
303 0 : }
304 2 : smallest = key.Clone()
305 : }
306 2 : if err := iter.Error(); err != nil {
307 0 : return nil, err
308 0 : }
309 2 : if s := iter.Last(); s != nil {
310 2 : k := s.SmallestKey()
311 2 : if err := ingestValidateKey(opts, &k); err != nil {
312 0 : return nil, err
313 0 : }
314 2 : largest := s.LargestKey().Clone()
315 2 : meta.ExtendPointKeyBounds(opts.Comparer.Compare, smallest, largest)
316 : }
317 : }
318 :
319 : // Update the range-key bounds for the table.
320 2 : {
321 2 : iter, err := r.NewRawRangeKeyIter()
322 2 : if err != nil {
323 0 : return nil, err
324 0 : }
325 2 : if iter != nil {
326 2 : defer iter.Close()
327 2 : var smallest InternalKey
328 2 : if s := iter.First(); s != nil {
329 2 : key := s.SmallestKey()
330 2 : if err := ingestValidateKey(opts, &key); err != nil {
331 0 : return nil, err
332 0 : }
333 2 : smallest = key.Clone()
334 : }
335 2 : if err := iter.Error(); err != nil {
336 0 : return nil, err
337 0 : }
338 2 : if s := iter.Last(); s != nil {
339 2 : k := s.SmallestKey()
340 2 : if err := ingestValidateKey(opts, &k); err != nil {
341 0 : return nil, err
342 0 : }
343 : // As range keys are fragmented, the end key of the last range key in
344 : // the table provides the upper bound for the table.
345 2 : largest := s.LargestKey().Clone()
346 2 : meta.ExtendRangeKeyBounds(opts.Comparer.Compare, smallest, largest)
347 : }
348 2 : if err := iter.Error(); err != nil {
349 0 : return nil, err
350 0 : }
351 : }
352 : }
353 :
354 2 : if !meta.HasPointKeys && !meta.HasRangeKeys {
355 2 : return nil, nil
356 2 : }
357 :
358 : // Sanity check that the various bounds on the file were set consistently.
359 2 : if err := meta.Validate(opts.Comparer.Compare, opts.Comparer.FormatKey); err != nil {
360 0 : return nil, err
361 0 : }
362 :
363 2 : return meta, nil
364 : }
365 :
366 : type ingestLoadResult struct {
367 : localMeta, sharedMeta []*fileMetadata
368 : externalMeta []*fileMetadata
369 : localPaths []string
370 : sharedLevels []uint8
371 : fileCount int
372 : }
373 :
374 : func ingestLoad(
375 : opts *Options,
376 : fmv FormatMajorVersion,
377 : paths []string,
378 : shared []SharedSSTMeta,
379 : external []ExternalFile,
380 : cacheID uint64,
381 : pending []base.DiskFileNum,
382 : objProvider objstorage.Provider,
383 : jobID int,
384 2 : ) (ingestLoadResult, error) {
385 2 : meta := make([]*fileMetadata, 0, len(paths))
386 2 : newPaths := make([]string, 0, len(paths))
387 2 : for i := range paths {
388 2 : f, err := opts.FS.Open(paths[i])
389 2 : if err != nil {
390 1 : return ingestLoadResult{}, err
391 1 : }
392 :
393 2 : readable, err := sstable.NewSimpleReadable(f)
394 2 : if err != nil {
395 1 : return ingestLoadResult{}, err
396 1 : }
397 2 : m, err := ingestLoad1(opts, fmv, readable, cacheID, pending[i])
398 2 : if err != nil {
399 1 : return ingestLoadResult{}, err
400 1 : }
401 2 : if m != nil {
402 2 : meta = append(meta, m)
403 2 : newPaths = append(newPaths, paths[i])
404 2 : }
405 : }
406 2 : if len(shared) == 0 && len(external) == 0 {
407 2 : return ingestLoadResult{localMeta: meta, localPaths: newPaths, fileCount: len(meta)}, nil
408 2 : }
409 :
410 : // Sort the shared files according to level.
411 2 : sort.Sort(sharedByLevel(shared))
412 2 :
413 2 : sharedMeta := make([]*fileMetadata, 0, len(shared))
414 2 : levels := make([]uint8, 0, len(shared))
415 2 : for i := range shared {
416 2 : m, err := ingestSynthesizeShared(opts, shared[i], pending[len(paths)+i])
417 2 : if err != nil {
418 0 : return ingestLoadResult{}, err
419 0 : }
420 2 : if shared[i].Level < sharedLevelsStart {
421 0 : return ingestLoadResult{}, errors.New("cannot ingest shared file in level below sharedLevelsStart")
422 0 : }
423 2 : sharedMeta = append(sharedMeta, m)
424 2 : levels = append(levels, shared[i].Level)
425 : }
426 2 : externalMeta := make([]*fileMetadata, 0, len(external))
427 2 : for i := range external {
428 1 : m, err := ingestLoad1External(opts, external[i], pending[len(paths)+len(shared)+i], objProvider, jobID)
429 1 : if err != nil {
430 0 : return ingestLoadResult{}, err
431 0 : }
432 1 : externalMeta = append(externalMeta, m)
433 : }
434 2 : result := ingestLoadResult{
435 2 : localMeta: meta,
436 2 : sharedMeta: sharedMeta,
437 2 : externalMeta: externalMeta,
438 2 : localPaths: newPaths,
439 2 : sharedLevels: levels,
440 2 : fileCount: len(meta) + len(sharedMeta) + len(externalMeta),
441 2 : }
442 2 : return result, nil
443 : }
444 :
445 : // Struct for sorting metadatas by smallest user keys, while ensuring the
446 : // matching path also gets swapped to the same index. For use in
447 : // ingestSortAndVerify.
448 : type metaAndPaths struct {
449 : meta []*fileMetadata
450 : paths []string
451 : cmp Compare
452 : }
453 :
454 2 : func (m metaAndPaths) Len() int {
455 2 : return len(m.meta)
456 2 : }
457 :
458 2 : func (m metaAndPaths) Less(i, j int) bool {
459 2 : return m.cmp(m.meta[i].Smallest.UserKey, m.meta[j].Smallest.UserKey) < 0
460 2 : }
461 :
462 2 : func (m metaAndPaths) Swap(i, j int) {
463 2 : m.meta[i], m.meta[j] = m.meta[j], m.meta[i]
464 2 : if m.paths != nil {
465 2 : m.paths[i], m.paths[j] = m.paths[j], m.paths[i]
466 2 : }
467 : }
468 :
469 2 : func ingestSortAndVerify(cmp Compare, lr ingestLoadResult, exciseSpan KeyRange) error {
470 2 : // Verify that all the shared files (i.e. files in sharedMeta)
471 2 : // fit within the exciseSpan.
472 2 : for i := range lr.sharedMeta {
473 2 : f := lr.sharedMeta[i]
474 2 : if !exciseSpan.Contains(cmp, f.Smallest) || !exciseSpan.Contains(cmp, f.Largest) {
475 0 : return errors.AssertionFailedf("pebble: shared file outside of excise span, span [%s-%s), file = %s", exciseSpan.Start, exciseSpan.End, f.String())
476 0 : }
477 : }
478 2 : if len(lr.externalMeta) > 0 {
479 1 : if len(lr.localMeta) > 0 || len(lr.sharedMeta) > 0 {
480 0 : // Currently we only support external ingests on their own. If external
481 0 : // files are present alongside local/shared files, return an error.
482 0 : return errors.AssertionFailedf("pebble: external files cannot be ingested atomically alongside other types of files")
483 0 : }
484 1 : sort.Sort(&metaAndPaths{
485 1 : meta: lr.externalMeta,
486 1 : cmp: cmp,
487 1 : })
488 1 : for i := 1; i < len(lr.externalMeta); i++ {
489 1 : if sstableKeyCompare(cmp, lr.externalMeta[i-1].Largest, lr.externalMeta[i].Smallest) >= 0 {
490 1 : return errors.AssertionFailedf("pebble: external sstables have overlapping ranges")
491 1 : }
492 : }
493 1 : return nil
494 : }
495 2 : if len(lr.localMeta) <= 1 || len(lr.localPaths) <= 1 {
496 2 : return nil
497 2 : }
498 :
499 2 : sort.Sort(&metaAndPaths{
500 2 : meta: lr.localMeta,
501 2 : paths: lr.localPaths,
502 2 : cmp: cmp,
503 2 : })
504 2 :
505 2 : for i := 1; i < len(lr.localPaths); i++ {
506 2 : if sstableKeyCompare(cmp, lr.localMeta[i-1].Largest, lr.localMeta[i].Smallest) >= 0 {
507 2 : return errors.AssertionFailedf("pebble: local ingestion sstables have overlapping ranges")
508 2 : }
509 : }
510 2 : if len(lr.sharedMeta) == 0 {
511 2 : return nil
512 2 : }
513 0 : filesInLevel := make([]*fileMetadata, 0, len(lr.sharedMeta))
514 0 : for l := sharedLevelsStart; l < numLevels; l++ {
515 0 : filesInLevel = filesInLevel[:0]
516 0 : for i := range lr.sharedMeta {
517 0 : if lr.sharedLevels[i] == uint8(l) {
518 0 : filesInLevel = append(filesInLevel, lr.sharedMeta[i])
519 0 : }
520 : }
521 0 : slices.SortFunc(filesInLevel, func(a, b *fileMetadata) int {
522 0 : return cmp(a.Smallest.UserKey, b.Smallest.UserKey)
523 0 : })
524 0 : for i := 1; i < len(filesInLevel); i++ {
525 0 : if sstableKeyCompare(cmp, filesInLevel[i-1].Largest, filesInLevel[i].Smallest) >= 0 {
526 0 : return errors.AssertionFailedf("pebble: external shared sstables have overlapping ranges")
527 0 : }
528 : }
529 : }
530 0 : return nil
531 : }
532 :
533 1 : func ingestCleanup(objProvider objstorage.Provider, meta []*fileMetadata) error {
534 1 : var firstErr error
535 1 : for i := range meta {
536 1 : if err := objProvider.Remove(fileTypeTable, meta[i].FileBacking.DiskFileNum); err != nil {
537 1 : firstErr = firstError(firstErr, err)
538 1 : }
539 : }
540 1 : return firstErr
541 : }
542 :
543 : // ingestLink creates new objects which are backed by either hardlinks to or
544 : // copies of the ingested files. It also attaches shared objects to the provider.
545 : func ingestLink(
546 : jobID int,
547 : opts *Options,
548 : objProvider objstorage.Provider,
549 : lr ingestLoadResult,
550 : shared []SharedSSTMeta,
551 2 : ) error {
552 2 : for i := range lr.localPaths {
553 2 : objMeta, err := objProvider.LinkOrCopyFromLocal(
554 2 : context.TODO(), opts.FS, lr.localPaths[i], fileTypeTable, lr.localMeta[i].FileBacking.DiskFileNum,
555 2 : objstorage.CreateOptions{PreferSharedStorage: true},
556 2 : )
557 2 : if err != nil {
558 1 : if err2 := ingestCleanup(objProvider, lr.localMeta[:i]); err2 != nil {
559 0 : opts.Logger.Errorf("ingest cleanup failed: %v", err2)
560 0 : }
561 1 : return err
562 : }
563 2 : if opts.EventListener.TableCreated != nil {
564 2 : opts.EventListener.TableCreated(TableCreateInfo{
565 2 : JobID: jobID,
566 2 : Reason: "ingesting",
567 2 : Path: objProvider.Path(objMeta),
568 2 : FileNum: lr.localMeta[i].FileNum,
569 2 : })
570 2 : }
571 : }
572 2 : sharedObjs := make([]objstorage.RemoteObjectToAttach, 0, len(shared))
573 2 : for i := range shared {
574 2 : backing, err := shared[i].Backing.Get()
575 2 : if err != nil {
576 0 : return err
577 0 : }
578 2 : sharedObjs = append(sharedObjs, objstorage.RemoteObjectToAttach{
579 2 : FileNum: lr.sharedMeta[i].FileBacking.DiskFileNum,
580 2 : FileType: fileTypeTable,
581 2 : Backing: backing,
582 2 : })
583 : }
584 2 : sharedObjMetas, err := objProvider.AttachRemoteObjects(sharedObjs)
585 2 : if err != nil {
586 0 : return err
587 0 : }
588 2 : for i := range sharedObjMetas {
589 2 : // One corner case around file sizes we need to be mindful of, is that
590 2 : // if one of the shareObjs was initially created by us (and has boomeranged
591 2 : // back from another node), we'll need to update the FileBacking's size
592 2 : // to be the true underlying size. Otherwise, we could hit errors when we
593 2 : // open the db again after a crash/restart (see checkConsistency in open.go),
594 2 : // plus it more accurately allows us to prioritize compactions of files
595 2 : // that were originally created by us.
596 2 : if sharedObjMetas[i].IsShared() && !objProvider.IsSharedForeign(sharedObjMetas[i]) {
597 2 : size, err := objProvider.Size(sharedObjMetas[i])
598 2 : if err != nil {
599 0 : return err
600 0 : }
601 2 : lr.sharedMeta[i].FileBacking.Size = uint64(size)
602 : }
603 2 : if opts.EventListener.TableCreated != nil {
604 2 : opts.EventListener.TableCreated(TableCreateInfo{
605 2 : JobID: jobID,
606 2 : Reason: "ingesting",
607 2 : Path: objProvider.Path(sharedObjMetas[i]),
608 2 : FileNum: lr.sharedMeta[i].FileNum,
609 2 : })
610 2 : }
611 : }
612 : // We do not need to do anything about lr.externalMetas. Those were already
613 : // linked in ingestLoad.
614 :
615 2 : return nil
616 : }
617 :
618 2 : func ingestMemtableOverlaps(cmp Compare, mem flushable, keyRanges []internalKeyRange) bool {
619 2 : iter := mem.newIter(nil)
620 2 : rangeDelIter := mem.newRangeDelIter(nil)
621 2 : rkeyIter := mem.newRangeKeyIter(nil)
622 2 :
623 2 : closeIters := func() error {
624 2 : err := iter.Close()
625 2 : if rangeDelIter != nil {
626 2 : err = firstError(err, rangeDelIter.Close())
627 2 : }
628 2 : if rkeyIter != nil {
629 2 : err = firstError(err, rkeyIter.Close())
630 2 : }
631 2 : return err
632 : }
633 :
634 2 : for _, kr := range keyRanges {
635 2 : if overlapWithIterator(iter, &rangeDelIter, rkeyIter, kr, cmp) {
636 2 : closeIters()
637 2 : return true
638 2 : }
639 : }
640 :
641 : // Assume overlap if any iterator errored out.
642 2 : return closeIters() != nil
643 : }
644 :
645 : func ingestUpdateSeqNum(
646 : cmp Compare, format base.FormatKey, seqNum uint64, loadResult ingestLoadResult,
647 2 : ) error {
648 2 : setSeqFn := func(k base.InternalKey) base.InternalKey {
649 2 : return base.MakeInternalKey(k.UserKey, seqNum, k.Kind())
650 2 : }
651 2 : updateMetadata := func(m *fileMetadata) error {
652 2 : // NB: we set the fields directly here, rather than via their Extend*
653 2 : // methods, as we are updating sequence numbers.
654 2 : if m.HasPointKeys {
655 2 : m.SmallestPointKey = setSeqFn(m.SmallestPointKey)
656 2 : }
657 2 : if m.HasRangeKeys {
658 2 : m.SmallestRangeKey = setSeqFn(m.SmallestRangeKey)
659 2 : }
660 2 : m.Smallest = setSeqFn(m.Smallest)
661 2 : // Only update the seqnum for the largest key if that key is not an
662 2 : // "exclusive sentinel" (i.e. a range deletion sentinel or a range key
663 2 : // boundary), as doing so effectively drops the exclusive sentinel (by
664 2 : // lowering the seqnum from the max value), and extends the bounds of the
665 2 : // table.
666 2 : // NB: as the largest range key is always an exclusive sentinel, it is never
667 2 : // updated.
668 2 : if m.HasPointKeys && !m.LargestPointKey.IsExclusiveSentinel() {
669 2 : m.LargestPointKey = setSeqFn(m.LargestPointKey)
670 2 : }
671 2 : if !m.Largest.IsExclusiveSentinel() {
672 2 : m.Largest = setSeqFn(m.Largest)
673 2 : }
674 : // Setting smallestSeqNum == largestSeqNum triggers the setting of
675 : // Properties.GlobalSeqNum when an sstable is loaded.
676 2 : m.SmallestSeqNum = seqNum
677 2 : m.LargestSeqNum = seqNum
678 2 : // Ensure the new bounds are consistent.
679 2 : if err := m.Validate(cmp, format); err != nil {
680 0 : return err
681 0 : }
682 2 : seqNum++
683 2 : return nil
684 : }
685 :
686 : // Shared sstables are required to be sorted by level ascending. We then
687 : // iterate the shared sstables in reverse, assigning the lower sequence
688 : // numbers to the shared sstables that will be ingested into the lower
689 : // (larger numbered) levels first. This ensures sequence number shadowing is
690 : // correct.
691 2 : for i := len(loadResult.sharedMeta) - 1; i >= 0; i-- {
692 2 : if i-1 >= 0 && loadResult.sharedLevels[i-1] > loadResult.sharedLevels[i] {
693 0 : panic(errors.AssertionFailedf("shared files %s, %s out of order", loadResult.sharedMeta[i-1], loadResult.sharedMeta[i]))
694 : }
695 2 : if err := updateMetadata(loadResult.sharedMeta[i]); err != nil {
696 0 : return err
697 0 : }
698 : }
699 2 : for i := range loadResult.localMeta {
700 2 : if err := updateMetadata(loadResult.localMeta[i]); err != nil {
701 0 : return err
702 0 : }
703 : }
704 2 : for i := range loadResult.externalMeta {
705 1 : if err := updateMetadata(loadResult.externalMeta[i]); err != nil {
706 0 : return err
707 0 : }
708 : }
709 2 : return nil
710 : }
711 :
712 : // Denotes an internal key range. Smallest and largest are both inclusive.
713 : type internalKeyRange struct {
714 : smallest, largest InternalKey
715 : }
716 :
717 : func overlapWithIterator(
718 : iter internalIterator,
719 : rangeDelIter *keyspan.FragmentIterator,
720 : rkeyIter keyspan.FragmentIterator,
721 : keyRange internalKeyRange,
722 : cmp Compare,
723 2 : ) bool {
724 2 : // Check overlap with point operations.
725 2 : //
726 2 : // When using levelIter, it seeks to the SST whose boundaries
727 2 : // contain keyRange.smallest.UserKey(S).
728 2 : // It then tries to find a point in that SST that is >= S.
729 2 : // If there's no such point it means the SST ends in a tombstone in which case
730 2 : // levelIter.SeekGE generates a boundary range del sentinel.
731 2 : // The comparison of this boundary with keyRange.largest(L) below
732 2 : // is subtle but maintains correctness.
733 2 : // 1) boundary < L,
734 2 : // since boundary is also > S (initial seek),
735 2 : // whatever the boundary's start key may be, we're always overlapping.
736 2 : // 2) boundary > L,
737 2 : // overlap with boundary cannot be determined since we don't know boundary's start key.
738 2 : // We require checking for overlap with rangeDelIter.
739 2 : // 3) boundary == L and L is not sentinel,
740 2 : // means boundary < L and hence is similar to 1).
741 2 : // 4) boundary == L and L is sentinel,
742 2 : // we'll always overlap since for any values of i,j ranges [i, k) and [j, k) always overlap.
743 2 : key, _ := iter.SeekGE(keyRange.smallest.UserKey, base.SeekGEFlagsNone)
744 2 : if key != nil {
745 2 : c := sstableKeyCompare(cmp, *key, keyRange.largest)
746 2 : if c <= 0 {
747 2 : return true
748 2 : }
749 : }
750 : // Assume overlap if iterator errored.
751 2 : if err := iter.Error(); err != nil {
752 1 : return true
753 1 : }
754 :
755 2 : computeOverlapWithSpans := func(rIter keyspan.FragmentIterator) bool {
756 2 : // NB: The spans surfaced by the fragment iterator are non-overlapping.
757 2 : span := rIter.SeekLT(keyRange.smallest.UserKey)
758 2 : if span == nil {
759 2 : span = rIter.Next()
760 2 : }
761 2 : for ; span != nil; span = rIter.Next() {
762 2 : if span.Empty() {
763 2 : continue
764 : }
765 2 : key := span.SmallestKey()
766 2 : c := sstableKeyCompare(cmp, key, keyRange.largest)
767 2 : if c > 0 {
768 2 : // The start of the span is after the largest key in the
769 2 : // ingested table.
770 2 : return false
771 2 : }
772 2 : if cmp(span.End, keyRange.smallest.UserKey) > 0 {
773 2 : // The end of the span is greater than the smallest in the
774 2 : // table. Note that the span end key is exclusive, thus ">0"
775 2 : // instead of ">=0".
776 2 : return true
777 2 : }
778 : }
779 : // Assume overlap if iterator errored.
780 2 : if err := rIter.Error(); err != nil {
781 0 : return true
782 0 : }
783 2 : return false
784 : }
785 :
786 : // rkeyIter is either a range key level iter, or a range key iterator
787 : // over a single file.
788 2 : if rkeyIter != nil {
789 2 : if computeOverlapWithSpans(rkeyIter) {
790 2 : return true
791 2 : }
792 : }
793 :
794 : // Check overlap with range deletions.
795 2 : if rangeDelIter == nil || *rangeDelIter == nil {
796 2 : return false
797 2 : }
798 2 : return computeOverlapWithSpans(*rangeDelIter)
799 : }
800 :
801 : // ingestTargetLevel returns the target level for a file being ingested.
802 : // If suggestSplit is true, it accounts for ingest-time splitting as part of
803 : // its target level calculation, and if a split candidate is found, that file
804 : // is returned as the splitFile.
805 : func ingestTargetLevel(
806 : newIters tableNewIters,
807 : newRangeKeyIter keyspan.TableNewSpanIter,
808 : iterOps IterOptions,
809 : comparer *Comparer,
810 : v *version,
811 : baseLevel int,
812 : compactions map[*compaction]struct{},
813 : meta *fileMetadata,
814 : suggestSplit bool,
815 2 : ) (targetLevel int, splitFile *fileMetadata, err error) {
816 2 : // Find the lowest level which does not have any files which overlap meta. We
817 2 : // search from L0 to L6 looking for whether there are any files in the level
818 2 : // which overlap meta. We want the "lowest" level (where lower means
819 2 : // increasing level number) in order to reduce write amplification.
820 2 : //
821 2 : // There are 2 kinds of overlap we need to check for: file boundary overlap
822 2 : // and data overlap. Data overlap implies file boundary overlap. Note that it
823 2 : // is always possible to ingest into L0.
824 2 : //
825 2 : // To place meta at level i where i > 0:
826 2 : // - there must not be any data overlap with levels <= i, since that will
827 2 : // violate the sequence number invariant.
828 2 : // - no file boundary overlap with level i, since that will violate the
829 2 : // invariant that files do not overlap in levels i > 0.
830 2 : // - if there is only a file overlap at a given level, and no data overlap,
831 2 : // we can still slot a file at that level. We return the fileMetadata with
832 2 : // which we have file boundary overlap (must be only one file, as sstable
833 2 : // bounds are usually tight on user keys) and the caller is expected to split
834 2 : // that sstable into two virtual sstables, allowing this file to go into that
835 2 : // level. Note that if we have file boundary overlap with two files, which
836 2 : // should only happen on rare occasions, we treat it as data overlap and
837 2 : // don't use this optimization.
838 2 : //
839 2 : // The file boundary overlap check is simpler to conceptualize. Consider the
840 2 : // following example, in which the ingested file lies completely before or
841 2 : // after the file being considered.
842 2 : //
843 2 : // |--| |--| ingested file: [a,b] or [f,g]
844 2 : // |-----| existing file: [c,e]
845 2 : // _____________________
846 2 : // a b c d e f g
847 2 : //
848 2 : // In both cases the ingested file can move to considering the next level.
849 2 : //
850 2 : // File boundary overlap does not necessarily imply data overlap. The check
851 2 : // for data overlap is a little more nuanced. Consider the following examples:
852 2 : //
853 2 : // 1. No data overlap:
854 2 : //
855 2 : // |-| |--| ingested file: [cc-d] or [ee-ff]
856 2 : // |*--*--*----*------*| existing file: [a-g], points: [a, b, c, dd, g]
857 2 : // _____________________
858 2 : // a b c d e f g
859 2 : //
860 2 : // In this case the ingested files can "fall through" this level. The checks
861 2 : // continue at the next level.
862 2 : //
863 2 : // 2. Data overlap:
864 2 : //
865 2 : // |--| ingested file: [d-e]
866 2 : // |*--*--*----*------*| existing file: [a-g], points: [a, b, c, dd, g]
867 2 : // _____________________
868 2 : // a b c d e f g
869 2 : //
870 2 : // In this case the file cannot be ingested into this level as the point 'dd'
871 2 : // is in the way.
872 2 : //
873 2 : // It is worth noting that the check for data overlap is only approximate. In
874 2 : // the previous example, the ingested table [d-e] could contain only the
875 2 : // points 'd' and 'e', in which case the table would be eligible for
876 2 : // considering lower levels. However, such a fine-grained check would need to
877 2 : // be exhaustive (comparing points and ranges in both the ingested existing
878 2 : // tables) and such a check is prohibitively expensive. Thus Pebble treats any
879 2 : // existing point that falls within the ingested table bounds as being "data
880 2 : // overlap".
881 2 :
882 2 : // This assertion implicitly checks that we have the current version of
883 2 : // the metadata.
884 2 : if v.L0Sublevels == nil {
885 0 : return 0, nil, errors.AssertionFailedf("could not read L0 sublevels")
886 0 : }
887 2 : iterOps.CategoryAndQoS = sstable.CategoryAndQoS{
888 2 : Category: "pebble-ingest",
889 2 : QoSLevel: sstable.LatencySensitiveQoSLevel,
890 2 : }
891 2 : // Check for overlap over the keys of L0 by iterating over the sublevels.
892 2 : for subLevel := 0; subLevel < len(v.L0SublevelFiles); subLevel++ {
893 2 : iter := newLevelIter(context.Background(),
894 2 : iterOps, comparer, newIters, v.L0Sublevels.Levels[subLevel].Iter(), manifest.Level(0), internalIterOpts{})
895 2 :
896 2 : var rangeDelIter keyspan.FragmentIterator
897 2 : // Pass in a non-nil pointer to rangeDelIter so that levelIter.findFileGE
898 2 : // sets it up for the target file.
899 2 : iter.initRangeDel(&rangeDelIter)
900 2 :
901 2 : levelIter := keyspan.LevelIter{}
902 2 : levelIter.Init(
903 2 : keyspan.SpanIterOptions{}, comparer.Compare, newRangeKeyIter,
904 2 : v.L0Sublevels.Levels[subLevel].Iter(), manifest.Level(0), manifest.KeyTypeRange,
905 2 : )
906 2 :
907 2 : kr := internalKeyRange{
908 2 : smallest: meta.Smallest,
909 2 : largest: meta.Largest,
910 2 : }
911 2 : overlap := overlapWithIterator(iter, &rangeDelIter, &levelIter, kr, comparer.Compare)
912 2 : err := iter.Close() // Closes range del iter as well.
913 2 : err = firstError(err, levelIter.Close())
914 2 : if err != nil {
915 1 : return 0, nil, err
916 1 : }
917 2 : if overlap {
918 2 : return targetLevel, nil, nil
919 2 : }
920 : }
921 :
922 2 : level := baseLevel
923 2 : for ; level < numLevels; level++ {
924 2 : levelIter := newLevelIter(context.Background(),
925 2 : iterOps, comparer, newIters, v.Levels[level].Iter(), manifest.Level(level), internalIterOpts{})
926 2 : var rangeDelIter keyspan.FragmentIterator
927 2 : // Pass in a non-nil pointer to rangeDelIter so that levelIter.findFileGE
928 2 : // sets it up for the target file.
929 2 : levelIter.initRangeDel(&rangeDelIter)
930 2 :
931 2 : rkeyLevelIter := &keyspan.LevelIter{}
932 2 : rkeyLevelIter.Init(
933 2 : keyspan.SpanIterOptions{}, comparer.Compare, newRangeKeyIter,
934 2 : v.Levels[level].Iter(), manifest.Level(level), manifest.KeyTypeRange,
935 2 : )
936 2 :
937 2 : kr := internalKeyRange{
938 2 : smallest: meta.Smallest,
939 2 : largest: meta.Largest,
940 2 : }
941 2 : overlap := overlapWithIterator(levelIter, &rangeDelIter, rkeyLevelIter, kr, comparer.Compare)
942 2 : err := levelIter.Close() // Closes range del iter as well.
943 2 : err = firstError(err, rkeyLevelIter.Close())
944 2 : if err != nil {
945 0 : return 0, nil, err
946 0 : }
947 2 : if overlap {
948 2 : return targetLevel, splitFile, nil
949 2 : }
950 :
951 : // Check boundary overlap.
952 2 : var candidateSplitFile *fileMetadata
953 2 : boundaryOverlaps := v.Overlaps(level, comparer.Compare, meta.Smallest.UserKey,
954 2 : meta.Largest.UserKey, meta.Largest.IsExclusiveSentinel())
955 2 : if !boundaryOverlaps.Empty() {
956 2 : // We are already guaranteed to not have any data overlaps with files
957 2 : // in boundaryOverlaps, otherwise we'd have returned in the above if
958 2 : // statements. Use this, plus boundaryOverlaps.Len() == 1 to detect for
959 2 : // the case where we can slot this file into the current level despite
960 2 : // a boundary overlap, by splitting one existing file into two virtual
961 2 : // sstables.
962 2 : if suggestSplit && boundaryOverlaps.Len() == 1 {
963 2 : iter := boundaryOverlaps.Iter()
964 2 : candidateSplitFile = iter.First()
965 2 : } else {
966 2 : // We either don't want to suggest ingest-time splits (i.e.
967 2 : // !suggestSplit), or we boundary-overlapped with more than one file.
968 2 : continue
969 : }
970 : }
971 :
972 : // Check boundary overlap with any ongoing compactions. We consider an
973 : // overlapping compaction that's writing files to an output level as
974 : // equivalent to boundary overlap with files in that output level.
975 : //
976 : // We cannot check for data overlap with the new SSTs compaction will produce
977 : // since compaction hasn't been done yet. However, there's no need to check
978 : // since all keys in them will be from levels in [c.startLevel,
979 : // c.outputLevel], and all those levels have already had their data overlap
980 : // tested negative (else we'd have returned earlier).
981 : //
982 : // An alternative approach would be to cancel these compactions and proceed
983 : // with an ingest-time split on this level if necessary. However, compaction
984 : // cancellation can result in significant wasted effort and is best avoided
985 : // unless necessary.
986 2 : overlaps := false
987 2 : for c := range compactions {
988 2 : if c.outputLevel == nil || level != c.outputLevel.level {
989 2 : continue
990 : }
991 2 : if comparer.Compare(meta.Smallest.UserKey, c.largest.UserKey) <= 0 &&
992 2 : comparer.Compare(meta.Largest.UserKey, c.smallest.UserKey) >= 0 {
993 2 : overlaps = true
994 2 : break
995 : }
996 : }
997 2 : if !overlaps {
998 2 : targetLevel = level
999 2 : splitFile = candidateSplitFile
1000 2 : }
1001 : }
1002 2 : return targetLevel, splitFile, nil
1003 : }
1004 :
1005 : // Ingest ingests a set of sstables into the DB. Ingestion of the files is
1006 : // atomic and semantically equivalent to creating a single batch containing all
1007 : // of the mutations in the sstables. Ingestion may require the memtable to be
1008 : // flushed. The ingested sstable files are moved into the DB and must reside on
1009 : // the same filesystem as the DB. Sstables can be created for ingestion using
1010 : // sstable.Writer. On success, Ingest removes the input paths.
1011 : //
1012 : // Two types of sstables are accepted for ingestion(s): one is sstables present
1013 : // in the instance's vfs.FS and can be referenced locally. The other is sstables
1014 : // present in remote.Storage, referred to as shared or foreign sstables. These
1015 : // shared sstables can be linked through objstorageprovider.Provider, and do not
1016 : // need to already be present on the local vfs.FS. Foreign sstables must all fit
1017 : // in an excise span, and are destined for a level specified in SharedSSTMeta.
1018 : //
1019 : // All sstables *must* be Sync()'d by the caller after all bytes are written
1020 : // and before its file handle is closed; failure to do so could violate
1021 : // durability or lead to corrupted on-disk state. This method cannot, in a
1022 : // platform-and-FS-agnostic way, ensure that all sstables in the input are
1023 : // properly synced to disk. Opening new file handles and Sync()-ing them
1024 : // does not always guarantee durability; see the discussion here on that:
1025 : // https://github.com/cockroachdb/pebble/pull/835#issuecomment-663075379
1026 : //
1027 : // Ingestion loads each sstable into the lowest level of the LSM which it
1028 : // doesn't overlap (see ingestTargetLevel). If an sstable overlaps a memtable,
1029 : // ingestion forces the memtable to flush, and then waits for the flush to
1030 : // occur. In some cases, such as with no foreign sstables and no excise span,
1031 : // ingestion that gets blocked on a memtable can join the flushable queue and
1032 : // finish even before the memtable has been flushed.
1033 : //
1034 : // The steps for ingestion are:
1035 : //
1036 : // 1. Allocate file numbers for every sstable being ingested.
1037 : // 2. Load the metadata for all sstables being ingested.
1038 : // 3. Sort the sstables by smallest key, verifying non overlap (for local
1039 : // sstables).
1040 : // 4. Hard link (or copy) the local sstables into the DB directory.
1041 : // 5. Allocate a sequence number to use for all of the entries in the
1042 : // local sstables. This is the step where overlap with memtables is
1043 : // determined. If there is overlap, we remember the most recent memtable
1044 : // that overlaps.
1045 : // 6. Update the sequence number in the ingested local sstables. (Remote
1046 : // sstables get fixed sequence numbers that were determined at load time.)
1047 : // 7. Wait for the most recent memtable that overlaps to flush (if any).
1048 : // 8. Add the ingested sstables to the version (DB.ingestApply).
1049 : // 8.1. If an excise span was specified, figure out what sstables in the
1050 : // current version overlap with the excise span, and create new virtual
1051 : // sstables out of those sstables that exclude the excised span (DB.excise).
1052 : // 9. Publish the ingestion sequence number.
1053 : //
1054 : // Note that if the mutable memtable overlaps with ingestion, a flush of the
1055 : // memtable is forced equivalent to DB.Flush. Additionally, subsequent
1056 : // mutations that get sequence numbers larger than the ingestion sequence
1057 : // number get queued up behind the ingestion waiting for it to complete. This
1058 : // can produce a noticeable hiccup in performance. See
1059 : // https://github.com/cockroachdb/pebble/issues/25 for an idea for how to fix
1060 : // this hiccup.
1061 2 : func (d *DB) Ingest(paths []string) error {
1062 2 : if err := d.closed.Load(); err != nil {
1063 1 : panic(err)
1064 : }
1065 2 : if d.opts.ReadOnly {
1066 1 : return ErrReadOnly
1067 1 : }
1068 2 : _, err := d.ingest(paths, ingestTargetLevel, nil /* shared */, KeyRange{}, nil /* external */)
1069 2 : return err
1070 : }
1071 :
1072 : // IngestOperationStats provides some information about where in the LSM the
1073 : // bytes were ingested.
1074 : type IngestOperationStats struct {
1075 : // Bytes is the total bytes in the ingested sstables.
1076 : Bytes uint64
1077 : // ApproxIngestedIntoL0Bytes is the approximate number of bytes ingested
1078 : // into L0. This value is approximate when flushable ingests are active and
1079 : // an ingest overlaps an entry in the flushable queue. Currently, this
1080 : // approximation is very rough, only including tables that overlapped the
1081 : // memtable. This estimate may be improved with #2112.
1082 : ApproxIngestedIntoL0Bytes uint64
1083 : // MemtableOverlappingFiles is the count of ingested sstables
1084 : // that overlapped keys in the memtables.
1085 : MemtableOverlappingFiles int
1086 : }
1087 :
1088 : // ExternalFile are external sstables that can be referenced through
1089 : // objprovider and ingested as remote files that will not be refcounted or
1090 : // cleaned up. For use with online restore. Note that the underlying sstable
1091 : // could contain keys outside the [Smallest,Largest) bounds; however Pebble
1092 : // is expected to only read the keys within those bounds.
1093 : type ExternalFile struct {
1094 : // Locator is the shared.Locator that can be used with objProvider to
1095 : // resolve a reference to this external sstable.
1096 : Locator remote.Locator
1097 : // ObjName is the unique name of this sstable on Locator.
1098 : ObjName string
1099 : // Size of the referenced proportion of the virtualized sstable. An estimate
1100 : // is acceptable in lieu of the backing file size.
1101 : Size uint64
1102 : // SmallestUserKey and LargestUserKey are the [smallest,largest) user key
1103 : // bounds of the sstable. Both these bounds are loose i.e. it's possible for
1104 : // the sstable to not span the entirety of this range. However, multiple
1105 : // ExternalFiles in one ingestion must all have non-overlapping
1106 : // [smallest, largest) spans. Note that this Largest bound is exclusive.
1107 : SmallestUserKey, LargestUserKey []byte
1108 : // HasPointKey and HasRangeKey denote whether this file contains point keys
1109 : // or range keys. If both structs are false, an error is returned during
1110 : // ingestion.
1111 : HasPointKey, HasRangeKey bool
1112 : }
1113 :
1114 : // IngestWithStats does the same as Ingest, and additionally returns
1115 : // IngestOperationStats.
1116 1 : func (d *DB) IngestWithStats(paths []string) (IngestOperationStats, error) {
1117 1 : if err := d.closed.Load(); err != nil {
1118 0 : panic(err)
1119 : }
1120 1 : if d.opts.ReadOnly {
1121 0 : return IngestOperationStats{}, ErrReadOnly
1122 0 : }
1123 1 : return d.ingest(paths, ingestTargetLevel, nil /* shared */, KeyRange{}, nil /* external */)
1124 : }
1125 :
1126 : // IngestExternalFiles does the same as IngestWithStats, and additionally
1127 : // accepts external files (with locator info that can be resolved using
1128 : // d.opts.SharedStorage). These files must also be non-overlapping with
1129 : // each other, and must be resolvable through d.objProvider.
1130 1 : func (d *DB) IngestExternalFiles(external []ExternalFile) (IngestOperationStats, error) {
1131 1 : if err := d.closed.Load(); err != nil {
1132 0 : panic(err)
1133 : }
1134 :
1135 1 : if d.opts.ReadOnly {
1136 0 : return IngestOperationStats{}, ErrReadOnly
1137 0 : }
1138 1 : if d.opts.Experimental.RemoteStorage == nil {
1139 0 : return IngestOperationStats{}, errors.New("pebble: cannot ingest external files without shared storage configured")
1140 0 : }
1141 1 : return d.ingest(nil, ingestTargetLevel, nil /* shared */, KeyRange{}, external)
1142 : }
1143 :
1144 : // IngestAndExcise does the same as IngestWithStats, and additionally accepts a
1145 : // list of shared files to ingest that can be read from a remote.Storage through
1146 : // a Provider. All the shared files must live within exciseSpan, and any existing
1147 : // keys in exciseSpan are deleted by turning existing sstables into virtual
1148 : // sstables (if not virtual already) and shrinking their spans to exclude
1149 : // exciseSpan. See the comment at Ingest for a more complete picture of the
1150 : // ingestion process.
1151 : //
1152 : // Panics if this DB instance was not instantiated with a remote.Storage and
1153 : // shared sstables are present.
1154 : func (d *DB) IngestAndExcise(
1155 : paths []string, shared []SharedSSTMeta, exciseSpan KeyRange,
1156 2 : ) (IngestOperationStats, error) {
1157 2 : if err := d.closed.Load(); err != nil {
1158 0 : panic(err)
1159 : }
1160 2 : if d.opts.ReadOnly {
1161 0 : return IngestOperationStats{}, ErrReadOnly
1162 0 : }
1163 2 : return d.ingest(paths, ingestTargetLevel, shared, exciseSpan, nil /* external */)
1164 : }
1165 :
1166 : // Both DB.mu and commitPipeline.mu must be held while this is called.
1167 : func (d *DB) newIngestedFlushableEntry(
1168 : meta []*fileMetadata, seqNum uint64, logNum base.DiskFileNum,
1169 2 : ) (*flushableEntry, error) {
1170 2 : // Update the sequence number for all of the sstables in the
1171 2 : // metadata. Writing the metadata to the manifest when the
1172 2 : // version edit is applied is the mechanism that persists the
1173 2 : // sequence number. The sstables themselves are left unmodified.
1174 2 : // In this case, a version edit will only be written to the manifest
1175 2 : // when the flushable is eventually flushed. If Pebble restarts in that
1176 2 : // time, then we'll lose the ingest sequence number information. But this
1177 2 : // information will also be reconstructed on node restart.
1178 2 : if err := ingestUpdateSeqNum(
1179 2 : d.cmp, d.opts.Comparer.FormatKey, seqNum, ingestLoadResult{localMeta: meta},
1180 2 : ); err != nil {
1181 0 : return nil, err
1182 0 : }
1183 :
1184 2 : f := newIngestedFlushable(meta, d.opts.Comparer, d.newIters, d.tableNewRangeKeyIter)
1185 2 :
1186 2 : // NB: The logNum/seqNum are the WAL number which we're writing this entry
1187 2 : // to and the sequence number within the WAL which we'll write this entry
1188 2 : // to.
1189 2 : entry := d.newFlushableEntry(f, logNum, seqNum)
1190 2 : // The flushable entry starts off with a single reader ref, so increment
1191 2 : // the FileMetadata.Refs.
1192 2 : for _, file := range f.files {
1193 2 : file.Ref()
1194 2 : }
1195 2 : entry.unrefFiles = func() []*fileBacking {
1196 2 : var obsolete []*fileBacking
1197 2 : for _, file := range f.files {
1198 2 : if file.Unref() == 0 {
1199 2 : obsolete = append(obsolete, file.FileMetadata.FileBacking)
1200 2 : }
1201 : }
1202 2 : return obsolete
1203 : }
1204 :
1205 2 : entry.flushForced = true
1206 2 : entry.releaseMemAccounting = func() {}
1207 2 : return entry, nil
1208 : }
1209 :
1210 : // Both DB.mu and commitPipeline.mu must be held while this is called. Since
1211 : // we're holding both locks, the order in which we rotate the memtable or
1212 : // recycle the WAL in this function is irrelevant as long as the correct log
1213 : // numbers are assigned to the appropriate flushable.
1214 2 : func (d *DB) handleIngestAsFlushable(meta []*fileMetadata, seqNum uint64) error {
1215 2 : b := d.NewBatch()
1216 2 : for _, m := range meta {
1217 2 : b.ingestSST(m.FileNum)
1218 2 : }
1219 2 : b.setSeqNum(seqNum)
1220 2 :
1221 2 : // If the WAL is disabled, then the logNum used to create the flushable
1222 2 : // entry doesn't matter. We just use the logNum assigned to the current
1223 2 : // mutable memtable. If the WAL is enabled, then this logNum will be
1224 2 : // overwritten by the logNum of the log which will contain the log entry
1225 2 : // for the ingestedFlushable.
1226 2 : logNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
1227 2 : if !d.opts.DisableWAL {
1228 2 : // We create a new WAL for the flushable instead of reusing the end of
1229 2 : // the previous WAL. This simplifies the increment of the minimum
1230 2 : // unflushed log number, and also simplifies WAL replay.
1231 2 : logNum, _ = d.recycleWAL()
1232 2 : d.mu.Unlock()
1233 2 : err := d.commit.directWrite(b)
1234 2 : if err != nil {
1235 0 : d.opts.Logger.Fatalf("%v", err)
1236 0 : }
1237 2 : d.mu.Lock()
1238 : }
1239 :
1240 2 : entry, err := d.newIngestedFlushableEntry(meta, seqNum, logNum)
1241 2 : if err != nil {
1242 0 : return err
1243 0 : }
1244 2 : nextSeqNum := seqNum + uint64(b.Count())
1245 2 :
1246 2 : // Set newLogNum to the logNum of the previous flushable. This value is
1247 2 : // irrelevant if the WAL is disabled. If the WAL is enabled, then we set
1248 2 : // the appropriate value below.
1249 2 : newLogNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
1250 2 : if !d.opts.DisableWAL {
1251 2 : // This is WAL num of the next mutable memtable which comes after the
1252 2 : // ingestedFlushable in the flushable queue. The mutable memtable
1253 2 : // will be created below.
1254 2 : newLogNum, _ = d.recycleWAL()
1255 2 : if err != nil {
1256 0 : return err
1257 0 : }
1258 : }
1259 :
1260 2 : currMem := d.mu.mem.mutable
1261 2 : // NB: Placing ingested sstables above the current memtables
1262 2 : // requires rotating of the existing memtables/WAL. There is
1263 2 : // some concern of churning through tiny memtables due to
1264 2 : // ingested sstables being placed on top of them, but those
1265 2 : // memtables would have to be flushed anyways.
1266 2 : d.mu.mem.queue = append(d.mu.mem.queue, entry)
1267 2 : d.rotateMemtable(newLogNum, nextSeqNum, currMem)
1268 2 : d.updateReadStateLocked(d.opts.DebugCheck)
1269 2 : d.maybeScheduleFlush()
1270 2 : return nil
1271 : }
1272 :
1273 : // See comment at Ingest() for details on how this works.
1274 : func (d *DB) ingest(
1275 : paths []string,
1276 : targetLevelFunc ingestTargetLevelFunc,
1277 : shared []SharedSSTMeta,
1278 : exciseSpan KeyRange,
1279 : external []ExternalFile,
1280 2 : ) (IngestOperationStats, error) {
1281 2 : if len(shared) > 0 && d.opts.Experimental.RemoteStorage == nil {
1282 0 : panic("cannot ingest shared sstables with nil SharedStorage")
1283 : }
1284 2 : if (exciseSpan.Valid() || len(shared) > 0 || len(external) > 0) && d.FormatMajorVersion() < FormatVirtualSSTables {
1285 0 : return IngestOperationStats{}, errors.New("pebble: format major version too old for excise, shared or external sstable ingestion")
1286 0 : }
1287 : // Allocate file numbers for all of the files being ingested and mark them as
1288 : // pending in order to prevent them from being deleted. Note that this causes
1289 : // the file number ordering to be out of alignment with sequence number
1290 : // ordering. The sorting of L0 tables by sequence number avoids relying on
1291 : // that (busted) invariant.
1292 2 : d.mu.Lock()
1293 2 : pendingOutputs := make([]base.DiskFileNum, len(paths)+len(shared)+len(external))
1294 2 : for i := 0; i < len(paths)+len(shared)+len(external); i++ {
1295 2 : pendingOutputs[i] = d.mu.versions.getNextDiskFileNum()
1296 2 : }
1297 :
1298 2 : jobID := d.mu.nextJobID
1299 2 : d.mu.nextJobID++
1300 2 : d.mu.Unlock()
1301 2 :
1302 2 : // Load the metadata for all the files being ingested. This step detects
1303 2 : // and elides empty sstables.
1304 2 : loadResult, err := ingestLoad(d.opts, d.FormatMajorVersion(), paths, shared, external, d.cacheID, pendingOutputs, d.objProvider, jobID)
1305 2 : if err != nil {
1306 1 : return IngestOperationStats{}, err
1307 1 : }
1308 :
1309 2 : if loadResult.fileCount == 0 {
1310 2 : // All of the sstables to be ingested were empty. Nothing to do.
1311 2 : return IngestOperationStats{}, nil
1312 2 : }
1313 :
1314 : // Verify the sstables do not overlap.
1315 2 : if err := ingestSortAndVerify(d.cmp, loadResult, exciseSpan); err != nil {
1316 2 : return IngestOperationStats{}, err
1317 2 : }
1318 :
1319 : // Hard link the sstables into the DB directory. Since the sstables aren't
1320 : // referenced by a version, they won't be used. If the hard linking fails
1321 : // (e.g. because the files reside on a different filesystem), ingestLink will
1322 : // fall back to copying, and if that fails we undo our work and return an
1323 : // error.
1324 2 : if err := ingestLink(jobID, d.opts, d.objProvider, loadResult, shared); err != nil {
1325 0 : return IngestOperationStats{}, err
1326 0 : }
1327 :
1328 : // Make the new tables durable. We need to do this at some point before we
1329 : // update the MANIFEST (via logAndApply), otherwise a crash can have the
1330 : // tables referenced in the MANIFEST, but not present in the provider.
1331 2 : if err := d.objProvider.Sync(); err != nil {
1332 1 : return IngestOperationStats{}, err
1333 1 : }
1334 :
1335 : // metaFlushableOverlaps is a slice parallel to meta indicating which of the
1336 : // ingested sstables overlap some table in the flushable queue. It's used to
1337 : // approximate ingest-into-L0 stats when using flushable ingests.
1338 2 : metaFlushableOverlaps := make([]bool, loadResult.fileCount)
1339 2 : var mem *flushableEntry
1340 2 : var mut *memTable
1341 2 : // asFlushable indicates whether the sstable was ingested as a flushable.
1342 2 : var asFlushable bool
1343 2 : iterOps := IterOptions{
1344 2 : CategoryAndQoS: sstable.CategoryAndQoS{
1345 2 : Category: "pebble-ingest",
1346 2 : QoSLevel: sstable.LatencySensitiveQoSLevel,
1347 2 : },
1348 2 : }
1349 2 : prepare := func(seqNum uint64) {
1350 2 : // Note that d.commit.mu is held by commitPipeline when calling prepare.
1351 2 :
1352 2 : d.mu.Lock()
1353 2 : defer d.mu.Unlock()
1354 2 :
1355 2 : // Check to see if any files overlap with any of the memtables. The queue
1356 2 : // is ordered from oldest to newest with the mutable memtable being the
1357 2 : // last element in the slice. We want to wait for the newest table that
1358 2 : // overlaps.
1359 2 :
1360 2 : for i := len(d.mu.mem.queue) - 1; i >= 0; i-- {
1361 2 : m := d.mu.mem.queue[i]
1362 2 : iter := m.newIter(&iterOps)
1363 2 : rangeDelIter := m.newRangeDelIter(&iterOps)
1364 2 : rkeyIter := m.newRangeKeyIter(&iterOps)
1365 2 :
1366 2 : checkForOverlap := func(i int, meta *fileMetadata) {
1367 2 : if metaFlushableOverlaps[i] {
1368 2 : // This table already overlapped a more recent flushable.
1369 2 : return
1370 2 : }
1371 2 : kr := internalKeyRange{
1372 2 : smallest: meta.Smallest,
1373 2 : largest: meta.Largest,
1374 2 : }
1375 2 : if overlapWithIterator(iter, &rangeDelIter, rkeyIter, kr, d.cmp) {
1376 2 : // If this is the first table to overlap a flushable, save
1377 2 : // the flushable. This ingest must be ingested or flushed
1378 2 : // after it.
1379 2 : if mem == nil {
1380 2 : mem = m
1381 2 : }
1382 2 : metaFlushableOverlaps[i] = true
1383 : }
1384 : }
1385 2 : for i := range loadResult.localMeta {
1386 2 : checkForOverlap(i, loadResult.localMeta[i])
1387 2 : }
1388 2 : for i := range loadResult.sharedMeta {
1389 2 : checkForOverlap(len(loadResult.localMeta)+i, loadResult.sharedMeta[i])
1390 2 : }
1391 2 : for i := range loadResult.externalMeta {
1392 1 : checkForOverlap(len(loadResult.localMeta)+len(loadResult.sharedMeta)+i, loadResult.externalMeta[i])
1393 1 : }
1394 2 : if exciseSpan.Valid() {
1395 2 : kr := internalKeyRange{
1396 2 : smallest: base.MakeInternalKey(exciseSpan.Start, InternalKeySeqNumMax, InternalKeyKindMax),
1397 2 : largest: base.MakeExclusiveSentinelKey(InternalKeyKindRangeDelete, exciseSpan.End),
1398 2 : }
1399 2 : if overlapWithIterator(iter, &rangeDelIter, rkeyIter, kr, d.cmp) {
1400 2 : if mem == nil {
1401 2 : mem = m
1402 2 : }
1403 : }
1404 : }
1405 2 : err := iter.Close()
1406 2 : if rangeDelIter != nil {
1407 2 : err = firstError(err, rangeDelIter.Close())
1408 2 : }
1409 2 : if rkeyIter != nil {
1410 2 : err = firstError(err, rkeyIter.Close())
1411 2 : }
1412 2 : if err != nil {
1413 0 : d.opts.Logger.Errorf("ingest error reading flushable for log %s: %s", m.logNum, err)
1414 0 : }
1415 : }
1416 :
1417 2 : if mem == nil {
1418 2 : // No overlap with any of the queued flushables, so no need to queue
1419 2 : // after them.
1420 2 :
1421 2 : // New writes with higher sequence numbers may be concurrently
1422 2 : // committed. We must ensure they don't flush before this ingest
1423 2 : // completes. To do that, we ref the mutable memtable as a writer,
1424 2 : // preventing its flushing (and the flushing of all subsequent
1425 2 : // flushables in the queue). Once we've acquired the manifest lock
1426 2 : // to add the ingested sstables to the LSM, we can unref as we're
1427 2 : // guaranteed that the flush won't edit the LSM before this ingest.
1428 2 : mut = d.mu.mem.mutable
1429 2 : mut.writerRef()
1430 2 : return
1431 2 : }
1432 : // The ingestion overlaps with some entry in the flushable queue.
1433 2 : if d.FormatMajorVersion() < FormatFlushableIngest ||
1434 2 : d.opts.Experimental.DisableIngestAsFlushable() ||
1435 2 : len(shared) > 0 || exciseSpan.Valid() || len(external) > 0 ||
1436 2 : (len(d.mu.mem.queue) > d.opts.MemTableStopWritesThreshold-1) {
1437 2 : // We're not able to ingest as a flushable,
1438 2 : // so we must synchronously flush.
1439 2 : //
1440 2 : // TODO(bilal): Currently, if any of the files being ingested are shared or
1441 2 : // there's an excise span present, we cannot use flushable ingests and need
1442 2 : // to wait synchronously. Either remove this caveat by fleshing out
1443 2 : // flushable ingest logic to also account for these cases, or remove this
1444 2 : // comment. Tracking issue: https://github.com/cockroachdb/pebble/issues/2676
1445 2 : if mem.flushable == d.mu.mem.mutable {
1446 2 : err = d.makeRoomForWrite(nil)
1447 2 : }
1448 : // New writes with higher sequence numbers may be concurrently
1449 : // committed. We must ensure they don't flush before this ingest
1450 : // completes. To do that, we ref the mutable memtable as a writer,
1451 : // preventing its flushing (and the flushing of all subsequent
1452 : // flushables in the queue). Once we've acquired the manifest lock
1453 : // to add the ingested sstables to the LSM, we can unref as we're
1454 : // guaranteed that the flush won't edit the LSM before this ingest.
1455 2 : mut = d.mu.mem.mutable
1456 2 : mut.writerRef()
1457 2 : mem.flushForced = true
1458 2 : d.maybeScheduleFlush()
1459 2 : return
1460 : }
1461 : // Since there aren't too many memtables already queued up, we can
1462 : // slide the ingested sstables on top of the existing memtables.
1463 2 : asFlushable = true
1464 2 : err = d.handleIngestAsFlushable(loadResult.localMeta, seqNum)
1465 : }
1466 :
1467 2 : var ve *versionEdit
1468 2 : apply := func(seqNum uint64) {
1469 2 : if err != nil || asFlushable {
1470 2 : // An error occurred during prepare.
1471 2 : if mut != nil {
1472 0 : if mut.writerUnref() {
1473 0 : d.mu.Lock()
1474 0 : d.maybeScheduleFlush()
1475 0 : d.mu.Unlock()
1476 0 : }
1477 : }
1478 2 : return
1479 : }
1480 :
1481 : // Update the sequence numbers for all ingested sstables'
1482 : // metadata. When the version edit is applied, the metadata is
1483 : // written to the manifest, persisting the sequence number.
1484 : // The sstables themselves are left unmodified.
1485 2 : if err = ingestUpdateSeqNum(
1486 2 : d.cmp, d.opts.Comparer.FormatKey, seqNum, loadResult,
1487 2 : ); err != nil {
1488 0 : if mut != nil {
1489 0 : if mut.writerUnref() {
1490 0 : d.mu.Lock()
1491 0 : d.maybeScheduleFlush()
1492 0 : d.mu.Unlock()
1493 0 : }
1494 : }
1495 0 : return
1496 : }
1497 :
1498 : // If we overlapped with a memtable in prepare wait for the flush to
1499 : // finish.
1500 2 : if mem != nil {
1501 2 : <-mem.flushed
1502 2 : }
1503 :
1504 : // Assign the sstables to the correct level in the LSM and apply the
1505 : // version edit.
1506 2 : ve, err = d.ingestApply(jobID, loadResult, targetLevelFunc, mut, exciseSpan)
1507 : }
1508 :
1509 : // Only one ingest can occur at a time because if not, one would block waiting
1510 : // for the other to finish applying. This blocking would happen while holding
1511 : // the commit mutex which would prevent unrelated batches from writing their
1512 : // changes to the WAL and memtable. This will cause a bigger commit hiccup
1513 : // during ingestion.
1514 2 : d.commit.ingestSem <- struct{}{}
1515 2 : d.commit.AllocateSeqNum(loadResult.fileCount, prepare, apply)
1516 2 : <-d.commit.ingestSem
1517 2 :
1518 2 : if err != nil {
1519 1 : if err2 := ingestCleanup(d.objProvider, loadResult.localMeta); err2 != nil {
1520 0 : d.opts.Logger.Errorf("ingest cleanup failed: %v", err2)
1521 0 : }
1522 2 : } else {
1523 2 : // Since we either created a hard link to the ingesting files, or copied
1524 2 : // them over, it is safe to remove the originals paths.
1525 2 : for _, path := range loadResult.localPaths {
1526 2 : if err2 := d.opts.FS.Remove(path); err2 != nil {
1527 1 : d.opts.Logger.Errorf("ingest failed to remove original file: %s", err2)
1528 1 : }
1529 : }
1530 : }
1531 :
1532 2 : info := TableIngestInfo{
1533 2 : JobID: jobID,
1534 2 : Err: err,
1535 2 : flushable: asFlushable,
1536 2 : }
1537 2 : if len(loadResult.localMeta) > 0 {
1538 2 : info.GlobalSeqNum = loadResult.localMeta[0].SmallestSeqNum
1539 2 : } else if len(loadResult.sharedMeta) > 0 {
1540 2 : info.GlobalSeqNum = loadResult.sharedMeta[0].SmallestSeqNum
1541 2 : } else {
1542 1 : info.GlobalSeqNum = loadResult.externalMeta[0].SmallestSeqNum
1543 1 : }
1544 2 : var stats IngestOperationStats
1545 2 : if ve != nil {
1546 2 : info.Tables = make([]struct {
1547 2 : TableInfo
1548 2 : Level int
1549 2 : }, len(ve.NewFiles))
1550 2 : for i := range ve.NewFiles {
1551 2 : e := &ve.NewFiles[i]
1552 2 : info.Tables[i].Level = e.Level
1553 2 : info.Tables[i].TableInfo = e.Meta.TableInfo()
1554 2 : stats.Bytes += e.Meta.Size
1555 2 : if e.Level == 0 {
1556 2 : stats.ApproxIngestedIntoL0Bytes += e.Meta.Size
1557 2 : }
1558 2 : if i < len(metaFlushableOverlaps) && metaFlushableOverlaps[i] {
1559 2 : stats.MemtableOverlappingFiles++
1560 2 : }
1561 : }
1562 2 : } else if asFlushable {
1563 2 : // NB: If asFlushable == true, there are no shared sstables.
1564 2 : info.Tables = make([]struct {
1565 2 : TableInfo
1566 2 : Level int
1567 2 : }, len(loadResult.localMeta))
1568 2 : for i, f := range loadResult.localMeta {
1569 2 : info.Tables[i].Level = -1
1570 2 : info.Tables[i].TableInfo = f.TableInfo()
1571 2 : stats.Bytes += f.Size
1572 2 : // We don't have exact stats on which files will be ingested into
1573 2 : // L0, because actual ingestion into the LSM has been deferred until
1574 2 : // flush time. Instead, we infer based on memtable overlap.
1575 2 : //
1576 2 : // TODO(jackson): If we optimistically compute data overlap (#2112)
1577 2 : // before entering the commit pipeline, we can use that overlap to
1578 2 : // improve our approximation by incorporating overlap with L0, not
1579 2 : // just memtables.
1580 2 : if metaFlushableOverlaps[i] {
1581 2 : stats.ApproxIngestedIntoL0Bytes += f.Size
1582 2 : stats.MemtableOverlappingFiles++
1583 2 : }
1584 : }
1585 : }
1586 2 : d.opts.EventListener.TableIngested(info)
1587 2 :
1588 2 : return stats, err
1589 : }
1590 :
1591 : // excise updates ve to include a replacement of the file m with new virtual
1592 : // sstables that exclude exciseSpan, returning a slice of newly-created files if
1593 : // any. If the entirety of m is deleted by exciseSpan, no new sstables are added
1594 : // and m is deleted. Note that ve is updated in-place.
1595 : //
1596 : // The manifest lock must be held when calling this method.
1597 : func (d *DB) excise(
1598 : exciseSpan KeyRange, m *fileMetadata, ve *versionEdit, level int,
1599 2 : ) ([]manifest.NewFileEntry, error) {
1600 2 : numCreatedFiles := 0
1601 2 : // Check if there's actually an overlap between m and exciseSpan.
1602 2 : if !exciseSpan.Overlaps(d.cmp, m) {
1603 2 : return nil, nil
1604 2 : }
1605 2 : ve.DeletedFiles[deletedFileEntry{
1606 2 : Level: level,
1607 2 : FileNum: m.FileNum,
1608 2 : }] = m
1609 2 : // Fast path: m sits entirely within the exciseSpan, so just delete it.
1610 2 : if exciseSpan.Contains(d.cmp, m.Smallest) && exciseSpan.Contains(d.cmp, m.Largest) {
1611 2 : return nil, nil
1612 2 : }
1613 2 : var iter internalIterator
1614 2 : var rangeDelIter keyspan.FragmentIterator
1615 2 : var rangeKeyIter keyspan.FragmentIterator
1616 2 : needsBacking := false
1617 2 : // Create a file to the left of the excise span, if necessary.
1618 2 : // The bounds of this file will be [m.Smallest, lastKeyBefore(exciseSpan.Start)].
1619 2 : //
1620 2 : // We create bounds that are tight on user keys, and we make the effort to find
1621 2 : // the last key in the original sstable that's smaller than exciseSpan.Start
1622 2 : // even though it requires some sstable reads. We could choose to create
1623 2 : // virtual sstables on loose userKey bounds, in which case we could just set
1624 2 : // leftFile.Largest to an exclusive sentinel at exciseSpan.Start. The biggest
1625 2 : // issue with that approach would be that it'd lead to lots of small virtual
1626 2 : // sstables in the LSM that have no guarantee on containing even a single user
1627 2 : // key within the file bounds. This has the potential to increase both read and
1628 2 : // write-amp as we will be opening up these sstables only to find no relevant
1629 2 : // keys in the read path, and compacting sstables on top of them instead of
1630 2 : // directly into the space occupied by them. We choose to incur the cost of
1631 2 : // calculating tight bounds at this time instead of creating more work in the
1632 2 : // future.
1633 2 : //
1634 2 : // TODO(bilal): Some of this work can happen without grabbing the manifest
1635 2 : // lock; we could grab one currentVersion, release the lock, calculate excised
1636 2 : // files, then grab the lock again and recalculate for just the files that
1637 2 : // have changed since our previous calculation. Do this optimiaztino as part of
1638 2 : // https://github.com/cockroachdb/pebble/issues/2112 .
1639 2 : if d.cmp(m.Smallest.UserKey, exciseSpan.Start) < 0 {
1640 2 : leftFile := &fileMetadata{
1641 2 : Virtual: true,
1642 2 : FileBacking: m.FileBacking,
1643 2 : FileNum: d.mu.versions.getNextFileNum(),
1644 2 : // Note that these are loose bounds for smallest/largest seqnums, but they're
1645 2 : // sufficient for maintaining correctness.
1646 2 : SmallestSeqNum: m.SmallestSeqNum,
1647 2 : LargestSeqNum: m.LargestSeqNum,
1648 2 : }
1649 2 : if m.HasPointKeys && !exciseSpan.Contains(d.cmp, m.SmallestPointKey) {
1650 2 : // This file will contain point keys
1651 2 : smallestPointKey := m.SmallestPointKey
1652 2 : var err error
1653 2 : iter, rangeDelIter, err = d.newIters(context.TODO(), m, &IterOptions{
1654 2 : CategoryAndQoS: sstable.CategoryAndQoS{
1655 2 : Category: "pebble-ingest",
1656 2 : QoSLevel: sstable.LatencySensitiveQoSLevel,
1657 2 : },
1658 2 : level: manifest.Level(level),
1659 2 : }, internalIterOpts{})
1660 2 : if err != nil {
1661 0 : return nil, err
1662 0 : }
1663 2 : var key *InternalKey
1664 2 : if iter != nil {
1665 2 : defer iter.Close()
1666 2 : key, _ = iter.SeekLT(exciseSpan.Start, base.SeekLTFlagsNone)
1667 2 : } else {
1668 0 : iter = emptyIter
1669 0 : }
1670 2 : if key != nil {
1671 2 : leftFile.ExtendPointKeyBounds(d.cmp, smallestPointKey, key.Clone())
1672 2 : }
1673 : // Store the min of (exciseSpan.Start, rdel.End) in lastRangeDel. This
1674 : // needs to be a copy if the key is owned by the range del iter.
1675 2 : var lastRangeDel []byte
1676 2 : if rangeDelIter != nil {
1677 2 : defer rangeDelIter.Close()
1678 2 : rdel := rangeDelIter.SeekLT(exciseSpan.Start)
1679 2 : if rdel != nil {
1680 2 : lastRangeDel = append(lastRangeDel[:0], rdel.End...)
1681 2 : if d.cmp(lastRangeDel, exciseSpan.Start) > 0 {
1682 2 : lastRangeDel = exciseSpan.Start
1683 2 : }
1684 : }
1685 2 : } else {
1686 2 : rangeDelIter = emptyKeyspanIter
1687 2 : }
1688 2 : if lastRangeDel != nil {
1689 2 : leftFile.ExtendPointKeyBounds(d.cmp, smallestPointKey, base.MakeExclusiveSentinelKey(InternalKeyKindRangeDelete, lastRangeDel))
1690 2 : }
1691 : }
1692 2 : if m.HasRangeKeys && !exciseSpan.Contains(d.cmp, m.SmallestRangeKey) {
1693 2 : // This file will contain range keys
1694 2 : var err error
1695 2 : smallestRangeKey := m.SmallestRangeKey
1696 2 : rangeKeyIter, err = d.tableNewRangeKeyIter(m, keyspan.SpanIterOptions{})
1697 2 : if err != nil {
1698 0 : return nil, err
1699 0 : }
1700 : // Store the min of (exciseSpan.Start, rkey.End) in lastRangeKey. This
1701 : // needs to be a copy if the key is owned by the range key iter.
1702 2 : var lastRangeKey []byte
1703 2 : var lastRangeKeyKind InternalKeyKind
1704 2 : defer rangeKeyIter.Close()
1705 2 : rkey := rangeKeyIter.SeekLT(exciseSpan.Start)
1706 2 : if rkey != nil {
1707 2 : lastRangeKey = append(lastRangeKey[:0], rkey.End...)
1708 2 : if d.cmp(lastRangeKey, exciseSpan.Start) > 0 {
1709 1 : lastRangeKey = exciseSpan.Start
1710 1 : }
1711 2 : lastRangeKeyKind = rkey.Keys[0].Kind()
1712 : }
1713 2 : if lastRangeKey != nil {
1714 2 : leftFile.ExtendRangeKeyBounds(d.cmp, smallestRangeKey, base.MakeExclusiveSentinelKey(lastRangeKeyKind, lastRangeKey))
1715 2 : }
1716 : }
1717 2 : if leftFile.HasRangeKeys || leftFile.HasPointKeys {
1718 2 : var err error
1719 2 : leftFile.Size, err = d.tableCache.estimateSize(m, leftFile.Smallest.UserKey, leftFile.Largest.UserKey)
1720 2 : if err != nil {
1721 0 : return nil, err
1722 0 : }
1723 2 : if leftFile.Size == 0 {
1724 2 : // On occasion, estimateSize gives us a low estimate, i.e. a 0 file size,
1725 2 : // such as if the excised file only has range keys/dels and no point
1726 2 : // keys. This can cause panics in places where we divide by file sizes.
1727 2 : // Correct for it here.
1728 2 : leftFile.Size = 1
1729 2 : }
1730 2 : if err := leftFile.Validate(d.cmp, d.opts.Comparer.FormatKey); err != nil {
1731 0 : return nil, err
1732 0 : }
1733 2 : leftFile.ValidateVirtual(m)
1734 2 : ve.NewFiles = append(ve.NewFiles, newFileEntry{Level: level, Meta: leftFile})
1735 2 : needsBacking = true
1736 2 : numCreatedFiles++
1737 : }
1738 : }
1739 : // Create a file to the right, if necessary.
1740 2 : if exciseSpan.Contains(d.cmp, m.Largest) {
1741 2 : // No key exists to the right of the excise span in this file.
1742 2 : if needsBacking && !m.Virtual {
1743 2 : // If m is virtual, then its file backing is already known to the manifest.
1744 2 : // We don't need to create another file backing. Note that there must be
1745 2 : // only one CreatedBackingTables entry per backing sstable. This is
1746 2 : // indicated by the VersionEdit.CreatedBackingTables invariant.
1747 2 : ve.CreatedBackingTables = append(ve.CreatedBackingTables, m.FileBacking)
1748 2 : }
1749 2 : return ve.NewFiles[len(ve.NewFiles)-numCreatedFiles:], nil
1750 : }
1751 : // Create a new file, rightFile, between [firstKeyAfter(exciseSpan.End), m.Largest].
1752 : //
1753 : // See comment before the definition of leftFile for the motivation behind
1754 : // calculating tight user-key bounds.
1755 2 : rightFile := &fileMetadata{
1756 2 : Virtual: true,
1757 2 : FileBacking: m.FileBacking,
1758 2 : FileNum: d.mu.versions.getNextFileNum(),
1759 2 : // Note that these are loose bounds for smallest/largest seqnums, but they're
1760 2 : // sufficient for maintaining correctness.
1761 2 : SmallestSeqNum: m.SmallestSeqNum,
1762 2 : LargestSeqNum: m.LargestSeqNum,
1763 2 : }
1764 2 : if m.HasPointKeys && !exciseSpan.Contains(d.cmp, m.LargestPointKey) {
1765 2 : // This file will contain point keys
1766 2 : largestPointKey := m.LargestPointKey
1767 2 : var err error
1768 2 : if iter == nil && rangeDelIter == nil {
1769 2 : iter, rangeDelIter, err = d.newIters(context.TODO(), m, &IterOptions{
1770 2 : CategoryAndQoS: sstable.CategoryAndQoS{
1771 2 : Category: "pebble-ingest",
1772 2 : QoSLevel: sstable.LatencySensitiveQoSLevel,
1773 2 : },
1774 2 : level: manifest.Level(level),
1775 2 : }, internalIterOpts{})
1776 2 : if err != nil {
1777 0 : return nil, err
1778 0 : }
1779 2 : if iter != nil {
1780 2 : defer iter.Close()
1781 2 : } else {
1782 0 : iter = emptyIter
1783 0 : }
1784 2 : if rangeDelIter != nil {
1785 2 : defer rangeDelIter.Close()
1786 2 : } else {
1787 2 : rangeDelIter = emptyKeyspanIter
1788 2 : }
1789 : }
1790 2 : key, _ := iter.SeekGE(exciseSpan.End, base.SeekGEFlagsNone)
1791 2 : if key != nil {
1792 2 : rightFile.ExtendPointKeyBounds(d.cmp, key.Clone(), largestPointKey)
1793 2 : }
1794 : // Store the max of (exciseSpan.End, rdel.Start) in firstRangeDel. This
1795 : // needs to be a copy if the key is owned by the range del iter.
1796 2 : var firstRangeDel []byte
1797 2 : rdel := rangeDelIter.SeekGE(exciseSpan.End)
1798 2 : if rdel != nil {
1799 2 : firstRangeDel = append(firstRangeDel[:0], rdel.Start...)
1800 2 : if d.cmp(firstRangeDel, exciseSpan.End) < 0 {
1801 2 : firstRangeDel = exciseSpan.End
1802 2 : }
1803 : }
1804 2 : if firstRangeDel != nil {
1805 2 : smallestPointKey := rdel.SmallestKey()
1806 2 : smallestPointKey.UserKey = firstRangeDel
1807 2 : rightFile.ExtendPointKeyBounds(d.cmp, smallestPointKey, largestPointKey)
1808 2 : }
1809 : }
1810 2 : if m.HasRangeKeys && !exciseSpan.Contains(d.cmp, m.LargestRangeKey) {
1811 2 : // This file will contain range keys.
1812 2 : largestRangeKey := m.LargestRangeKey
1813 2 : if rangeKeyIter == nil {
1814 2 : var err error
1815 2 : rangeKeyIter, err = d.tableNewRangeKeyIter(m, keyspan.SpanIterOptions{})
1816 2 : if err != nil {
1817 0 : return nil, err
1818 0 : }
1819 2 : defer rangeKeyIter.Close()
1820 : }
1821 : // Store the max of (exciseSpan.End, rkey.Start) in firstRangeKey. This
1822 : // needs to be a copy if the key is owned by the range key iter.
1823 2 : var firstRangeKey []byte
1824 2 : rkey := rangeKeyIter.SeekGE(exciseSpan.End)
1825 2 : if rkey != nil {
1826 2 : firstRangeKey = append(firstRangeKey[:0], rkey.Start...)
1827 2 : if d.cmp(firstRangeKey, exciseSpan.End) < 0 {
1828 2 : firstRangeKey = exciseSpan.End
1829 2 : }
1830 : }
1831 2 : if firstRangeKey != nil {
1832 2 : smallestRangeKey := rkey.SmallestKey()
1833 2 : smallestRangeKey.UserKey = firstRangeKey
1834 2 : // We call ExtendRangeKeyBounds so any internal boundType fields are
1835 2 : // set correctly. Note that this is mildly wasteful as we'll be comparing
1836 2 : // rightFile.{Smallest,Largest}RangeKey with themselves, which can be
1837 2 : // avoided if we exported ExtendOverallKeyBounds or so.
1838 2 : rightFile.ExtendRangeKeyBounds(d.cmp, smallestRangeKey, largestRangeKey)
1839 2 : }
1840 : }
1841 2 : if rightFile.HasRangeKeys || rightFile.HasPointKeys {
1842 2 : var err error
1843 2 : rightFile.Size, err = d.tableCache.estimateSize(m, rightFile.Smallest.UserKey, rightFile.Largest.UserKey)
1844 2 : if err != nil {
1845 0 : return nil, err
1846 0 : }
1847 2 : if rightFile.Size == 0 {
1848 2 : // On occasion, estimateSize gives us a low estimate, i.e. a 0 file size,
1849 2 : // such as if the excised file only has range keys/dels and no point keys.
1850 2 : // This can cause panics in places where we divide by file sizes. Correct
1851 2 : // for it here.
1852 2 : rightFile.Size = 1
1853 2 : }
1854 2 : rightFile.ValidateVirtual(m)
1855 2 : ve.NewFiles = append(ve.NewFiles, newFileEntry{Level: level, Meta: rightFile})
1856 2 : needsBacking = true
1857 2 : numCreatedFiles++
1858 : }
1859 :
1860 2 : if needsBacking && !m.Virtual {
1861 2 : // If m is virtual, then its file backing is already known to the manifest.
1862 2 : // We don't need to create another file backing. Note that there must be
1863 2 : // only one CreatedBackingTables entry per backing sstable. This is
1864 2 : // indicated by the VersionEdit.CreatedBackingTables invariant.
1865 2 : ve.CreatedBackingTables = append(ve.CreatedBackingTables, m.FileBacking)
1866 2 : }
1867 :
1868 2 : if err := rightFile.Validate(d.cmp, d.opts.Comparer.FormatKey); err != nil {
1869 0 : return nil, err
1870 0 : }
1871 2 : return ve.NewFiles[len(ve.NewFiles)-numCreatedFiles:], nil
1872 : }
1873 :
1874 : type ingestTargetLevelFunc func(
1875 : newIters tableNewIters,
1876 : newRangeKeyIter keyspan.TableNewSpanIter,
1877 : iterOps IterOptions,
1878 : comparer *Comparer,
1879 : v *version,
1880 : baseLevel int,
1881 : compactions map[*compaction]struct{},
1882 : meta *fileMetadata,
1883 : suggestSplit bool,
1884 : ) (int, *fileMetadata, error)
1885 :
1886 : type ingestSplitFile struct {
1887 : // ingestFile is the file being ingested.
1888 : ingestFile *fileMetadata
1889 : // splitFile is the file that needs to be split to allow ingestFile to slot
1890 : // into `level` level.
1891 : splitFile *fileMetadata
1892 : // The level where ingestFile will go (and where splitFile already is).
1893 : level int
1894 : }
1895 :
1896 : // ingestSplit splits files specified in `files` and updates ve in-place to
1897 : // account for existing files getting split into two virtual sstables. The map
1898 : // `replacedFiles` contains an in-progress map of all files that have been
1899 : // replaced with new virtual sstables in this version edit so far, which is also
1900 : // updated in-place.
1901 : //
1902 : // d.mu as well as the manifest lock must be held when calling this method.
1903 : func (d *DB) ingestSplit(
1904 : ve *versionEdit,
1905 : updateMetrics func(*fileMetadata, int, []newFileEntry),
1906 : files []ingestSplitFile,
1907 : replacedFiles map[base.FileNum][]newFileEntry,
1908 2 : ) error {
1909 2 : for _, s := range files {
1910 2 : // replacedFiles can be thought of as a tree, where we start iterating with
1911 2 : // s.splitFile and run its fileNum through replacedFiles, then find which of
1912 2 : // the replaced files overlaps with s.ingestFile, which becomes the new
1913 2 : // splitFile, then we check splitFile's replacements in replacedFiles again
1914 2 : // for overlap with s.ingestFile, and so on until we either can't find the
1915 2 : // current splitFile in replacedFiles (i.e. that's the file that now needs to
1916 2 : // be split), or we don't find a file that overlaps with s.ingestFile, which
1917 2 : // means a prior ingest split already produced enough room for s.ingestFile
1918 2 : // to go into this level without necessitating another ingest split.
1919 2 : splitFile := s.splitFile
1920 2 : for splitFile != nil {
1921 2 : replaced, ok := replacedFiles[splitFile.FileNum]
1922 2 : if !ok {
1923 2 : break
1924 : }
1925 2 : updatedSplitFile := false
1926 2 : for i := range replaced {
1927 2 : if replaced[i].Meta.Overlaps(d.cmp, s.ingestFile.Smallest.UserKey, s.ingestFile.Largest.UserKey, s.ingestFile.Largest.IsExclusiveSentinel()) {
1928 2 : if updatedSplitFile {
1929 0 : // This should never happen because the earlier ingestTargetLevel
1930 0 : // function only finds split file candidates that are guaranteed to
1931 0 : // have no data overlap, only boundary overlap. See the comments
1932 0 : // in that method to see the definitions of data vs boundary
1933 0 : // overlap. That, plus the fact that files in `replaced` are
1934 0 : // guaranteed to have file bounds that are tight on user keys
1935 0 : // (as that's what `d.excise` produces), means that the only case
1936 0 : // where we overlap with two or more files in `replaced` is if we
1937 0 : // actually had data overlap all along, or if the ingestion files
1938 0 : // were overlapping, either of which is an invariant violation.
1939 0 : panic("updated with two files in ingestSplit")
1940 : }
1941 2 : splitFile = replaced[i].Meta
1942 2 : updatedSplitFile = true
1943 : }
1944 : }
1945 2 : if !updatedSplitFile {
1946 2 : // None of the replaced files overlapped with the file being ingested.
1947 2 : // This can happen if we've already excised a span overlapping with
1948 2 : // this file, or if we have consecutive ingested files that can slide
1949 2 : // within the same gap between keys in an existing file. For instance,
1950 2 : // if an existing file has keys a and g and we're ingesting b-c, d-e,
1951 2 : // the first loop iteration will split the existing file into one that
1952 2 : // ends in a and another that starts at g, and the second iteration will
1953 2 : // fall into this case and require no splitting.
1954 2 : //
1955 2 : // No splitting necessary.
1956 2 : splitFile = nil
1957 2 : }
1958 : }
1959 2 : if splitFile == nil {
1960 2 : continue
1961 : }
1962 : // NB: excise operates on [start, end). We're splitting at [start, end]
1963 : // (assuming !s.ingestFile.Largest.IsExclusiveSentinel()). The conflation
1964 : // of exclusive vs inclusive end bounds should not make a difference here
1965 : // as we're guaranteed to not have any data overlap between splitFile and
1966 : // s.ingestFile, so panic if we do see a newly added file with an endKey
1967 : // equalling s.ingestFile.Largest, and !s.ingestFile.Largest.IsExclusiveSentinel()
1968 2 : added, err := d.excise(KeyRange{Start: s.ingestFile.Smallest.UserKey, End: s.ingestFile.Largest.UserKey}, splitFile, ve, s.level)
1969 2 : if err != nil {
1970 0 : return err
1971 0 : }
1972 2 : if _, ok := ve.DeletedFiles[deletedFileEntry{
1973 2 : Level: s.level,
1974 2 : FileNum: splitFile.FileNum,
1975 2 : }]; !ok {
1976 0 : panic("did not split file that was expected to be split")
1977 : }
1978 2 : replacedFiles[splitFile.FileNum] = added
1979 2 : for i := range added {
1980 2 : if s.ingestFile.Overlaps(d.cmp, added[i].Meta.Smallest.UserKey, added[i].Meta.Largest.UserKey, added[i].Meta.Largest.IsExclusiveSentinel()) {
1981 0 : panic("ingest-time split produced a file that overlaps with ingested file")
1982 : }
1983 : }
1984 2 : updateMetrics(splitFile, s.level, added)
1985 : }
1986 : // Flatten the version edit by removing any entries from ve.NewFiles that
1987 : // are also in ve.DeletedFiles.
1988 2 : newNewFiles := ve.NewFiles[:0]
1989 2 : for i := range ve.NewFiles {
1990 2 : fn := ve.NewFiles[i].Meta.FileNum
1991 2 : deEntry := deletedFileEntry{Level: ve.NewFiles[i].Level, FileNum: fn}
1992 2 : if _, ok := ve.DeletedFiles[deEntry]; ok {
1993 2 : delete(ve.DeletedFiles, deEntry)
1994 2 : } else {
1995 2 : newNewFiles = append(newNewFiles, ve.NewFiles[i])
1996 2 : }
1997 : }
1998 2 : ve.NewFiles = newNewFiles
1999 2 : return nil
2000 : }
2001 :
2002 : func (d *DB) ingestApply(
2003 : jobID int,
2004 : lr ingestLoadResult,
2005 : findTargetLevel ingestTargetLevelFunc,
2006 : mut *memTable,
2007 : exciseSpan KeyRange,
2008 2 : ) (*versionEdit, error) {
2009 2 : d.mu.Lock()
2010 2 : defer d.mu.Unlock()
2011 2 :
2012 2 : ve := &versionEdit{
2013 2 : NewFiles: make([]newFileEntry, lr.fileCount),
2014 2 : }
2015 2 : if exciseSpan.Valid() || (d.opts.Experimental.IngestSplit != nil && d.opts.Experimental.IngestSplit()) {
2016 2 : ve.DeletedFiles = map[manifest.DeletedFileEntry]*manifest.FileMetadata{}
2017 2 : }
2018 2 : metrics := make(map[int]*LevelMetrics)
2019 2 :
2020 2 : // Lock the manifest for writing before we use the current version to
2021 2 : // determine the target level. This prevents two concurrent ingestion jobs
2022 2 : // from using the same version to determine the target level, and also
2023 2 : // provides serialization with concurrent compaction and flush jobs.
2024 2 : // logAndApply unconditionally releases the manifest lock, but any earlier
2025 2 : // returns must unlock the manifest.
2026 2 : d.mu.versions.logLock()
2027 2 :
2028 2 : if mut != nil {
2029 2 : // Unref the mutable memtable to allows its flush to proceed. Now that we've
2030 2 : // acquired the manifest lock, we can be certain that if the mutable
2031 2 : // memtable has received more recent conflicting writes, the flush won't
2032 2 : // beat us to applying to the manifest resulting in sequence number
2033 2 : // inversion. Even though we call maybeScheduleFlush right now, this flush
2034 2 : // will apply after our ingestion.
2035 2 : if mut.writerUnref() {
2036 2 : d.maybeScheduleFlush()
2037 2 : }
2038 : }
2039 :
2040 2 : shouldIngestSplit := d.opts.Experimental.IngestSplit != nil &&
2041 2 : d.opts.Experimental.IngestSplit() && d.FormatMajorVersion() >= FormatVirtualSSTables
2042 2 : current := d.mu.versions.currentVersion()
2043 2 : baseLevel := d.mu.versions.picker.getBaseLevel()
2044 2 : iterOps := IterOptions{logger: d.opts.Logger}
2045 2 : // filesToSplit is a list where each element is a pair consisting of a file
2046 2 : // being ingested and a file being split to make room for an ingestion into
2047 2 : // that level. Each ingested file will appear at most once in this list. It
2048 2 : // is possible for split files to appear twice in this list.
2049 2 : filesToSplit := make([]ingestSplitFile, 0)
2050 2 : checkCompactions := false
2051 2 : for i := 0; i < lr.fileCount; i++ {
2052 2 : // Determine the lowest level in the LSM for which the sstable doesn't
2053 2 : // overlap any existing files in the level.
2054 2 : var m *fileMetadata
2055 2 : sharedIdx := -1
2056 2 : sharedLevel := -1
2057 2 : externalFile := false
2058 2 : if i < len(lr.localMeta) {
2059 2 : // local file.
2060 2 : m = lr.localMeta[i]
2061 2 : } else if (i - len(lr.localMeta)) < len(lr.sharedMeta) {
2062 2 : // shared file.
2063 2 : sharedIdx = i - len(lr.localMeta)
2064 2 : m = lr.sharedMeta[sharedIdx]
2065 2 : sharedLevel = int(lr.sharedLevels[sharedIdx])
2066 2 : } else {
2067 1 : // external file.
2068 1 : externalFile = true
2069 1 : m = lr.externalMeta[i-(len(lr.localMeta)+len(lr.sharedMeta))]
2070 1 : }
2071 2 : f := &ve.NewFiles[i]
2072 2 : var err error
2073 2 : if sharedIdx >= 0 {
2074 2 : f.Level = sharedLevel
2075 2 : if f.Level < sharedLevelsStart {
2076 0 : panic("cannot slot a shared file higher than the highest shared level")
2077 : }
2078 2 : ve.CreatedBackingTables = append(ve.CreatedBackingTables, m.FileBacking)
2079 2 : } else {
2080 2 : if externalFile {
2081 1 : ve.CreatedBackingTables = append(ve.CreatedBackingTables, m.FileBacking)
2082 1 : }
2083 2 : var splitFile *fileMetadata
2084 2 : if exciseSpan.Valid() && exciseSpan.Contains(d.cmp, m.Smallest) && exciseSpan.Contains(d.cmp, m.Largest) {
2085 2 : // This file fits perfectly within the excise span. We can slot it at
2086 2 : // L6, or sharedLevelsStart - 1 if we have shared files.
2087 2 : if len(lr.sharedMeta) > 0 {
2088 2 : f.Level = sharedLevelsStart - 1
2089 2 : if baseLevel > f.Level {
2090 2 : f.Level = 0
2091 2 : }
2092 2 : } else {
2093 2 : f.Level = 6
2094 2 : }
2095 2 : } else {
2096 2 : // TODO(bilal): findTargetLevel does disk IO (reading files for data
2097 2 : // overlap) even though we're holding onto d.mu. Consider unlocking
2098 2 : // d.mu while we do this. We already hold versions.logLock so we should
2099 2 : // not see any version applications while we're at this. The one
2100 2 : // complication here would be pulling out the mu.compact.inProgress
2101 2 : // check from findTargetLevel, as that requires d.mu to be held.
2102 2 : f.Level, splitFile, err = findTargetLevel(
2103 2 : d.newIters, d.tableNewRangeKeyIter, iterOps, d.opts.Comparer, current, baseLevel, d.mu.compact.inProgress, m, shouldIngestSplit)
2104 2 : }
2105 :
2106 2 : if splitFile != nil {
2107 2 : if invariants.Enabled {
2108 2 : if lf := current.Levels[f.Level].Find(d.cmp, splitFile); lf == nil {
2109 0 : panic("splitFile returned is not in level it should be")
2110 : }
2111 : }
2112 : // We take advantage of the fact that we won't drop the db mutex
2113 : // between now and the call to logAndApply. So, no files should
2114 : // get added to a new in-progress compaction at this point. We can
2115 : // avoid having to iterate on in-progress compactions to cancel them
2116 : // if none of the files being split have a compacting state.
2117 2 : if splitFile.IsCompacting() {
2118 1 : checkCompactions = true
2119 1 : }
2120 2 : filesToSplit = append(filesToSplit, ingestSplitFile{ingestFile: m, splitFile: splitFile, level: f.Level})
2121 : }
2122 : }
2123 2 : if err != nil {
2124 1 : d.mu.versions.logUnlock()
2125 1 : return nil, err
2126 1 : }
2127 2 : f.Meta = m
2128 2 : levelMetrics := metrics[f.Level]
2129 2 : if levelMetrics == nil {
2130 2 : levelMetrics = &LevelMetrics{}
2131 2 : metrics[f.Level] = levelMetrics
2132 2 : }
2133 2 : levelMetrics.NumFiles++
2134 2 : levelMetrics.Size += int64(m.Size)
2135 2 : levelMetrics.BytesIngested += m.Size
2136 2 : levelMetrics.TablesIngested++
2137 : }
2138 : // replacedFiles maps files excised due to exciseSpan (or splitFiles returned
2139 : // by ingestTargetLevel), to files that were created to replace it. This map
2140 : // is used to resolve references to split files in filesToSplit, as it is
2141 : // possible for a file that we want to split to no longer exist or have a
2142 : // newer fileMetadata due to a split induced by another ingestion file, or an
2143 : // excise.
2144 2 : replacedFiles := make(map[base.FileNum][]newFileEntry)
2145 2 : updateLevelMetricsOnExcise := func(m *fileMetadata, level int, added []newFileEntry) {
2146 2 : levelMetrics := metrics[level]
2147 2 : if levelMetrics == nil {
2148 2 : levelMetrics = &LevelMetrics{}
2149 2 : metrics[level] = levelMetrics
2150 2 : }
2151 2 : levelMetrics.NumFiles--
2152 2 : levelMetrics.Size -= int64(m.Size)
2153 2 : for i := range added {
2154 2 : levelMetrics.NumFiles++
2155 2 : levelMetrics.Size += int64(added[i].Meta.Size)
2156 2 : }
2157 : }
2158 2 : if exciseSpan.Valid() {
2159 2 : // Iterate through all levels and find files that intersect with exciseSpan.
2160 2 : //
2161 2 : // TODO(bilal): We could drop the DB mutex here as we don't need it for
2162 2 : // excises; we only need to hold the version lock which we already are
2163 2 : // holding. However releasing the DB mutex could mess with the
2164 2 : // ingestTargetLevel calculation that happened above, as it assumed that it
2165 2 : // had a complete view of in-progress compactions that wouldn't change
2166 2 : // until logAndApply is called. If we were to drop the mutex now, we could
2167 2 : // schedule another in-progress compaction that would go into the chosen target
2168 2 : // level and lead to file overlap within level (which would panic in
2169 2 : // logAndApply). We should drop the db mutex here, do the excise, then
2170 2 : // re-grab the DB mutex and rerun just the in-progress compaction check to
2171 2 : // see if any new compactions are conflicting with our chosen target levels
2172 2 : // for files, and if they are, we should signal those compactions to error
2173 2 : // out.
2174 2 : for level := range current.Levels {
2175 2 : overlaps := current.Overlaps(level, d.cmp, exciseSpan.Start, exciseSpan.End, true /* exclusiveEnd */)
2176 2 : iter := overlaps.Iter()
2177 2 :
2178 2 : for m := iter.First(); m != nil; m = iter.Next() {
2179 2 : newFiles, err := d.excise(exciseSpan, m, ve, level)
2180 2 : if err != nil {
2181 0 : return nil, err
2182 0 : }
2183 :
2184 2 : if _, ok := ve.DeletedFiles[deletedFileEntry{
2185 2 : Level: level,
2186 2 : FileNum: m.FileNum,
2187 2 : }]; !ok {
2188 2 : // We did not excise this file.
2189 2 : continue
2190 : }
2191 2 : replacedFiles[m.FileNum] = newFiles
2192 2 : updateLevelMetricsOnExcise(m, level, newFiles)
2193 : }
2194 : }
2195 : }
2196 2 : if len(filesToSplit) > 0 {
2197 2 : // For the same reasons as the above call to excise, we hold the db mutex
2198 2 : // while calling this method.
2199 2 : if err := d.ingestSplit(ve, updateLevelMetricsOnExcise, filesToSplit, replacedFiles); err != nil {
2200 0 : return nil, err
2201 0 : }
2202 : }
2203 2 : if len(filesToSplit) > 0 || exciseSpan.Valid() {
2204 2 : for c := range d.mu.compact.inProgress {
2205 2 : if c.versionEditApplied {
2206 1 : continue
2207 : }
2208 : // Check if this compaction overlaps with the excise span. Note that just
2209 : // checking if the inputs individually overlap with the excise span
2210 : // isn't sufficient; for instance, a compaction could have [a,b] and [e,f]
2211 : // as inputs and write it all out as [a,b,e,f] in one sstable. If we're
2212 : // doing a [c,d) excise at the same time as this compaction, we will have
2213 : // to error out the whole compaction as we can't guarantee it hasn't/won't
2214 : // write a file overlapping with the excise span.
2215 2 : if exciseSpan.OverlapsInternalKeyRange(d.cmp, c.smallest, c.largest) {
2216 2 : c.cancel.Store(true)
2217 2 : }
2218 : // Check if this compaction's inputs have been replaced due to an
2219 : // ingest-time split. In that case, cancel the compaction as a newly picked
2220 : // compaction would need to include any new files that slid in between
2221 : // previously-existing files. Note that we cancel any compaction that has a
2222 : // file that was ingest-split as an input, even if it started before this
2223 : // ingestion.
2224 2 : if checkCompactions {
2225 1 : for i := range c.inputs {
2226 1 : iter := c.inputs[i].files.Iter()
2227 1 : for f := iter.First(); f != nil; f = iter.Next() {
2228 1 : if _, ok := replacedFiles[f.FileNum]; ok {
2229 1 : c.cancel.Store(true)
2230 1 : break
2231 : }
2232 : }
2233 : }
2234 : }
2235 : }
2236 : // Check for any EventuallyFileOnlySnapshots that could be watching for
2237 : // an excise on this span.
2238 2 : if exciseSpan.Valid() {
2239 2 : for s := d.mu.snapshots.root.next; s != &d.mu.snapshots.root; s = s.next {
2240 2 : if s.efos == nil {
2241 0 : continue
2242 : }
2243 2 : efos := s.efos
2244 2 : // TODO(bilal): We can make this faster by taking advantage of the sorted
2245 2 : // nature of protectedRanges to do a sort.Search, or even maintaining a
2246 2 : // global list of all protected ranges instead of having to peer into every
2247 2 : // snapshot.
2248 2 : for i := range efos.protectedRanges {
2249 2 : if efos.protectedRanges[i].OverlapsKeyRange(d.cmp, exciseSpan) {
2250 2 : efos.excised.Store(true)
2251 2 : break
2252 : }
2253 : }
2254 : }
2255 : }
2256 : }
2257 2 : if err := d.mu.versions.logAndApply(jobID, ve, metrics, false /* forceRotation */, func() []compactionInfo {
2258 2 : return d.getInProgressCompactionInfoLocked(nil)
2259 2 : }); err != nil {
2260 1 : return nil, err
2261 1 : }
2262 :
2263 2 : d.mu.versions.metrics.Ingest.Count++
2264 2 :
2265 2 : d.updateReadStateLocked(d.opts.DebugCheck)
2266 2 : // updateReadStateLocked could have generated obsolete tables, schedule a
2267 2 : // cleanup job if necessary.
2268 2 : d.deleteObsoleteFiles(jobID)
2269 2 : d.updateTableStatsLocked(ve.NewFiles)
2270 2 : // The ingestion may have pushed a level over the threshold for compaction,
2271 2 : // so check to see if one is necessary and schedule it.
2272 2 : d.maybeScheduleCompaction()
2273 2 : var toValidate []manifest.NewFileEntry
2274 2 : dedup := make(map[base.DiskFileNum]struct{})
2275 2 : for _, entry := range ve.NewFiles {
2276 2 : if _, ok := dedup[entry.Meta.FileBacking.DiskFileNum]; !ok {
2277 2 : toValidate = append(toValidate, entry)
2278 2 : dedup[entry.Meta.FileBacking.DiskFileNum] = struct{}{}
2279 2 : }
2280 : }
2281 2 : d.maybeValidateSSTablesLocked(toValidate)
2282 2 : return ve, nil
2283 : }
2284 :
2285 : // maybeValidateSSTablesLocked adds the slice of newFileEntrys to the pending
2286 : // queue of files to be validated, when the feature is enabled.
2287 : //
2288 : // Note that if two entries with the same backing file are added twice, then the
2289 : // block checksums for the backing file will be validated twice.
2290 : //
2291 : // DB.mu must be locked when calling.
2292 2 : func (d *DB) maybeValidateSSTablesLocked(newFiles []newFileEntry) {
2293 2 : // Only add to the validation queue when the feature is enabled.
2294 2 : if !d.opts.Experimental.ValidateOnIngest {
2295 2 : return
2296 2 : }
2297 :
2298 2 : d.mu.tableValidation.pending = append(d.mu.tableValidation.pending, newFiles...)
2299 2 : if d.shouldValidateSSTablesLocked() {
2300 2 : go d.validateSSTables()
2301 2 : }
2302 : }
2303 :
2304 : // shouldValidateSSTablesLocked returns true if SSTable validation should run.
2305 : // DB.mu must be locked when calling.
2306 2 : func (d *DB) shouldValidateSSTablesLocked() bool {
2307 2 : return !d.mu.tableValidation.validating &&
2308 2 : d.closed.Load() == nil &&
2309 2 : d.opts.Experimental.ValidateOnIngest &&
2310 2 : len(d.mu.tableValidation.pending) > 0
2311 2 : }
2312 :
2313 : // validateSSTables runs a round of validation on the tables in the pending
2314 : // queue.
2315 2 : func (d *DB) validateSSTables() {
2316 2 : d.mu.Lock()
2317 2 : if !d.shouldValidateSSTablesLocked() {
2318 2 : d.mu.Unlock()
2319 2 : return
2320 2 : }
2321 :
2322 2 : pending := d.mu.tableValidation.pending
2323 2 : d.mu.tableValidation.pending = nil
2324 2 : d.mu.tableValidation.validating = true
2325 2 : jobID := d.mu.nextJobID
2326 2 : d.mu.nextJobID++
2327 2 : rs := d.loadReadState()
2328 2 :
2329 2 : // Drop DB.mu before performing IO.
2330 2 : d.mu.Unlock()
2331 2 :
2332 2 : // Validate all tables in the pending queue. This could lead to a situation
2333 2 : // where we are starving IO from other tasks due to having to page through
2334 2 : // all the blocks in all the sstables in the queue.
2335 2 : // TODO(travers): Add some form of pacing to avoid IO starvation.
2336 2 :
2337 2 : // If we fail to validate any files due to reasons other than uncovered
2338 2 : // corruption, accumulate them and re-queue them for another attempt.
2339 2 : var retry []manifest.NewFileEntry
2340 2 :
2341 2 : for _, f := range pending {
2342 2 : // The file may have been moved or deleted since it was ingested, in
2343 2 : // which case we skip.
2344 2 : if !rs.current.Contains(f.Level, d.cmp, f.Meta) {
2345 1 : // Assume the file was moved to a lower level. It is rare enough
2346 1 : // that a table is moved or deleted between the time it was ingested
2347 1 : // and the time the validation routine runs that the overall cost of
2348 1 : // this inner loop is tolerably low, when amortized over all
2349 1 : // ingested tables.
2350 1 : found := false
2351 1 : for i := f.Level + 1; i < numLevels; i++ {
2352 1 : if rs.current.Contains(i, d.cmp, f.Meta) {
2353 1 : found = true
2354 1 : break
2355 : }
2356 : }
2357 1 : if !found {
2358 1 : continue
2359 : }
2360 : }
2361 :
2362 2 : var err error
2363 2 : if f.Meta.Virtual {
2364 1 : err = d.tableCache.withVirtualReader(
2365 1 : f.Meta.VirtualMeta(), func(v sstable.VirtualReader) error {
2366 1 : return v.ValidateBlockChecksumsOnBacking()
2367 1 : })
2368 2 : } else {
2369 2 : err = d.tableCache.withReader(
2370 2 : f.Meta.PhysicalMeta(), func(r *sstable.Reader) error {
2371 2 : return r.ValidateBlockChecksums()
2372 2 : })
2373 : }
2374 :
2375 2 : if err != nil {
2376 1 : if IsCorruptionError(err) {
2377 1 : // TODO(travers): Hook into the corruption reporting pipeline, once
2378 1 : // available. See pebble#1192.
2379 1 : d.opts.Logger.Fatalf("pebble: encountered corruption during ingestion: %s", err)
2380 1 : } else {
2381 1 : // If there was some other, possibly transient, error that
2382 1 : // caused table validation to fail inform the EventListener and
2383 1 : // move on. We remember the table so that we can retry it in a
2384 1 : // subsequent table validation job.
2385 1 : //
2386 1 : // TODO(jackson): If the error is not transient, this will retry
2387 1 : // validation indefinitely. While not great, it's the same
2388 1 : // behavior as erroring flushes and compactions. We should
2389 1 : // address this as a part of #270.
2390 1 : d.opts.EventListener.BackgroundError(err)
2391 1 : retry = append(retry, f)
2392 1 : continue
2393 : }
2394 : }
2395 :
2396 2 : d.opts.EventListener.TableValidated(TableValidatedInfo{
2397 2 : JobID: jobID,
2398 2 : Meta: f.Meta,
2399 2 : })
2400 : }
2401 2 : rs.unref()
2402 2 : d.mu.Lock()
2403 2 : defer d.mu.Unlock()
2404 2 : d.mu.tableValidation.pending = append(d.mu.tableValidation.pending, retry...)
2405 2 : d.mu.tableValidation.validating = false
2406 2 : d.mu.tableValidation.cond.Broadcast()
2407 2 : if d.shouldValidateSSTablesLocked() {
2408 2 : go d.validateSSTables()
2409 2 : }
2410 : }
|