db: tolerate empty external ingestions

We now tolerate empty external ingestions.

When we encounter an empty table in a download compaction, we just
delete the table.

The iterator correctness issues that occurred before seem to no longer
apply, very likely thanks to the recent refactoring and cleanup of the
iterator stack.

Author: RaduBerinde

compaction.go

@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/compact"
-	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -2163,14 +2162,26 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 // d.mu must be held when calling this method. The mutex will be released when
 // doing IO.
 func (d *DB) runCopyCompaction(
-	jobID JobID,
-	c *compaction,
-	inputMeta *fileMetadata,
-	objMeta objstorage.ObjectMetadata,
-	ve *versionEdit,
-) error {
-	ctx := context.TODO()
+	jobID JobID, c *compaction,
+) (ve *versionEdit, stats compact.Stats, _ error) {
+	iter := c.startLevel.files.Iter()
+	inputMeta := iter.First()
+	if iter.Next() != nil {
+		return nil, compact.Stats{}, base.AssertionFailedf("got more than one file for a move compaction")
+	}
+	if c.cancel.Load() {
+		return nil, compact.Stats{}, ErrCancelledCompaction
+	}
+	ve = &versionEdit{
+		DeletedFiles: map[deletedFileEntry]*fileMetadata{
+			{Level: c.startLevel.level, FileNum: inputMeta.FileNum}: inputMeta,
+		},
+	}
 
+	objMeta, err := d.objProvider.Lookup(fileTypeTable, inputMeta.FileBacking.DiskFileNum)
+	if err != nil {
+		return nil, compact.Stats{}, err
+	}
 	if !objMeta.IsExternal() {
 		if objMeta.IsRemote() || !remote.ShouldCreateShared(d.opts.Experimental.CreateOnShared, c.outputLevel.level) {
 			panic("pebble: scheduled a copy compaction that is not actually moving files to shared storage")
@@ -2215,14 +2226,6 @@ func (d *DB) runCopyCompaction(
 		newMeta.InitPhysicalBacking()
 	}
 
-	c.metrics = map[int]*LevelMetrics{
-		c.outputLevel.level: {
-			BytesIn:         inputMeta.Size,
-			BytesCompacted:  inputMeta.Size,
-			TablesCompacted: 1,
-		},
-	}
-
 	// Before dropping the db mutex, grab a ref to the current version. This
 	// prevents any concurrent excises from deleting files that this compaction
 	// needs to read/maintain a reference to.
@@ -2244,11 +2247,12 @@ func (d *DB) runCopyCompaction(
 
 	// If the src obj is external, we're doing an external to local/shared copy.
 	if objMeta.IsExternal() {
+		ctx := context.TODO()
 		src, err := d.objProvider.OpenForReading(
 			ctx, fileTypeTable, inputMeta.FileBacking.DiskFileNum, objstorage.OpenOptions{},
 		)
 		if err != nil {
-			return err
+			return nil, compact.Stats{}, err
 		}
 		defer func() {
 			if src != nil {
@@ -2263,7 +2267,7 @@ func (d *DB) runCopyCompaction(
 			},
 		)
 		if err != nil {
-			return err
+			return nil, compact.Stats{}, err
 		}
 		deleteOnExit = true
 
@@ -2287,30 +2291,49 @@ func (d *DB) runCopyCompaction(
 		)
 		src = nil // We passed src to CopySpan; it's responsible for closing it.
 		if err != nil {
-			return err
+			if errors.Is(err, sstable.ErrEmptySpan) {
+				// The virtual table was empty. Just remove the backing file.
+				// Note that deleteOnExit is true so we will delete the created object.
+				c.metrics = map[int]*LevelMetrics{
+					c.outputLevel.level: {
+						BytesIn: inputMeta.Size,
+					},
+				}
+				return ve, compact.Stats{}, nil
+			}
+			return nil, compact.Stats{}, err
 		}
 		newMeta.FileBacking.Size = wrote
 		newMeta.Size = wrote
 	} else {
 		_, err := d.objProvider.LinkOrCopyFromLocal(context.TODO(), d.opts.FS,
 			d.objProvider.Path(objMeta), fileTypeTable, newMeta.FileBacking.DiskFileNum,
 			objstorage.CreateOptions{PreferSharedStorage: true})
-
 		if err != nil {
-			return err
+			return nil, compact.Stats{}, err
 		}
 		deleteOnExit = true
 	}
-	ve.NewFiles[0].Meta = newMeta
+	ve.NewFiles = []newFileEntry{{
+		Level: c.outputLevel.level,
+		Meta:  newMeta,
+	}}
 	if newMeta.Virtual {
 		ve.CreatedBackingTables = []*fileBacking{newMeta.FileBacking}
 	}
+	c.metrics = map[int]*LevelMetrics{
+		c.outputLevel.level: {
+			BytesIn:         inputMeta.Size,
+			BytesCompacted:  newMeta.Size,
+			TablesCompacted: 1,
+		},
+	}
 
 	if err := d.objProvider.Sync(); err != nil {
-		return err
+		return nil, compact.Stats{}, err
 	}
 	deleteOnExit = false
-	return nil
+	return ve, compact.Stats{}, nil
 }
 
 func (d *DB) runDeleteOnlyCompaction(
@@ -2334,23 +2357,17 @@ func (d *DB) runDeleteOnlyCompaction(
 	return ve, stats, nil
 }
 
-func (d *DB) runMoveOrCopyCompaction(
+func (d *DB) runMoveCompaction(
 	jobID JobID, c *compaction,
 ) (ve *versionEdit, stats compact.Stats, _ error) {
 	iter := c.startLevel.files.Iter()
 	meta := iter.First()
-	if invariants.Enabled {
-		if iter.Next() != nil {
-			panic("got more than one file for a move or copy compaction")
-		}
+	if iter.Next() != nil {
+		return nil, stats, base.AssertionFailedf("got more than one file for a move compaction")
 	}
 	if c.cancel.Load() {
 		return ve, stats, ErrCancelledCompaction
 	}
-	objMeta, err := d.objProvider.Lookup(fileTypeTable, meta.FileBacking.DiskFileNum)
-	if err != nil {
-		return ve, stats, err
-	}
 	c.metrics = map[int]*LevelMetrics{
 		c.outputLevel.level: {
 			BytesMoved:  meta.Size,
@@ -2365,12 +2382,8 @@ func (d *DB) runMoveOrCopyCompaction(
 			{Level: c.outputLevel.level, Meta: meta},
 		},
 	}
-	if c.kind == compactionKindMove {
-		return ve, stats, nil
-	}
 
-	err = d.runCopyCompaction(jobID, c, meta, objMeta, ve)
-	return ve, stats, err
+	return ve, stats, nil
 }
 
 // runCompaction runs a compaction that produces new on-disk tables from
@@ -2386,8 +2399,10 @@ func (d *DB) runCompaction(
 	switch c.kind {
 	case compactionKindDeleteOnly:
 		return d.runDeleteOnlyCompaction(jobID, c)
-	case compactionKindMove, compactionKindCopy:
-		return d.runMoveOrCopyCompaction(jobID, c)
+	case compactionKindMove:
+		return d.runMoveCompaction(jobID, c)
+	case compactionKindCopy:
+		return d.runCopyCompaction(jobID, c)
 	case compactionKindIngestedFlushable:
 		panic("pebble: runCompaction cannot handle compactionKindIngestedFlushable.")
 	}

ingest.go

@@ -698,68 +698,6 @@ func (d *DB) ingestAttachRemote(jobID JobID, lr ingestLoadResult) error {
 		}
 	}
 
-	if invariants.Enabled || d.opts.Experimental.CheckExternalIngestions {
-		// Verify that the ingestion is not empty. Empty tables cause correctness
-		// issues in some iterators.
-		for i := range lr.external {
-			e := &lr.external[i]
-			isEmpty, err := func() (bool, error) {
-				readable, err := d.objProvider.OpenForReading(
-					context.Background(), fileTypeTable, e.FileBacking.DiskFileNum, objstorage.OpenOptions{},
-				)
-				if err != nil {
-					return false, err
-				}
-				reader, err := sstable.NewReader(readable, d.opts.MakeReaderOptions())
-				if err != nil {
-					readable.Close()
-					return false, err
-				}
-				defer reader.Close()
-				iter, err := reader.NewIter(e.IterTransforms(), nil, nil)
-				if err != nil {
-					return false, err
-				}
-				if iter != nil {
-					defer iter.Close()
-					if kv := iter.SeekGE(e.external.StartKey, base.SeekGEFlagsNone); kv != nil {
-						cmp := d.opts.Comparer.Compare(kv.K.UserKey, e.external.EndKey)
-						if cmp < 0 || (cmp == 0 && e.external.EndKeyIsInclusive) {
-							return false, nil
-						}
-					}
-					if err := iter.Error(); err != nil {
-						return false, err
-					}
-				}
-				rangeIter, err := reader.NewRawRangeDelIter(e.IterTransforms())
-				if err != nil {
-					return false, err
-				}
-				if rangeIter != nil {
-					defer rangeIter.Close()
-					span, err := rangeIter.SeekGE(e.external.StartKey)
-					if err != nil {
-						return false, err
-					}
-					if span != nil {
-						cmp := d.opts.Comparer.Compare(span.Start, e.external.EndKey)
-						if cmp < 0 || (cmp == 0 && e.external.EndKeyIsInclusive) {
-							return false, nil
-						}
-					}
-				}
-				return true, nil
-			}()
-			if err != nil {
-				return err
-			}
-			if isEmpty {
-				panic(fmt.Sprintf("external ingestion is empty: %s (%q %q)", e.external.ObjName, e.external.StartKey, e.external.EndKey))
-			}
-		}
-	}
-
 	if d.opts.EventListener.TableCreated != nil {
 		for i := range remoteObjMetas {
 			d.opts.EventListener.TableCreated(TableCreateInfo{

metamorphic/build.go

@@ -274,32 +274,6 @@ func openExternalObj(
 	return reader, pointIter, rangeDelIter, rangeKeyIter
 }
 
-// externalObjIsEmpty returns true if the given external object has no point or
-// range keys withing the given bounds.
-func externalObjIsEmpty(
-	t *Test, externalObjID objID, bounds pebble.KeyRange, syntheticPrefix sstable.SyntheticPrefix,
-) bool {
-	reader, pointIter, rangeDelIter, rangeKeyIter := openExternalObj(t, externalObjID, bounds, syntheticPrefix)
-	defer reader.Close()
-	defer closeIters(pointIter, rangeDelIter, rangeKeyIter)
-
-	kv := pointIter.First()
-	panicIfErr(pointIter.Error())
-	if kv != nil {
-		return false
-	}
-	for _, it := range []keyspan.FragmentIterator{rangeDelIter, rangeKeyIter} {
-		if it != nil {
-			span, err := it.First()
-			panicIfErr(err)
-			if span != nil {
-				return false
-			}
-		}
-	}
-	return true
-}
-
 func panicIfErr(err error) {
 	if err != nil {
 		panic(err)

metamorphic/ops.go

@@ -995,33 +995,12 @@ type externalObjWithBounds struct {
 func (o *ingestExternalFilesOp) run(t *Test, h historyRecorder) {
 	db := t.getDB(o.dbID)
 
-	// We modify objs to eliminate empty objects.
-	var objs []externalObjWithBounds
-	for _, obj := range o.objs {
-		// Make sure the object exists and is not empty.
-		objMeta := t.getExternalObj(obj.externalObjID)
-		if m := objMeta.sstMeta; !m.HasPointKeys && !m.HasRangeKeys && !m.HasRangeDelKeys {
-			continue
-		}
-		if externalObjIsEmpty(t, obj.externalObjID, obj.bounds, obj.syntheticPrefix) {
-			// Currently we don't support ingesting external objects that have no point
-			// or range keys in the given range. Filter out any such objects.
-			// TODO(radu): even though we don't expect this case in practice, eventually
-			// we want to make sure that it doesn't cause failures.
-			continue
-		}
-		objs = append(objs, obj)
-	}
-	if len(objs) == 0 {
-		h.Recordf("%s // no-op", o)
-		return
-	}
 	var err error
 	if !t.testOpts.externalStorageEnabled {
 		// Emulate the operation by crating local, truncated SST files and ingesting
 		// them.
 		var paths []string
-		for i, obj := range objs {
+		for i, obj := range o.objs {
 			// Make sure the object exists and is not empty.
 			path, sstMeta := buildForIngestExternalEmulation(
 				t, o.dbID, obj.externalObjID, obj.bounds, obj.syntheticSuffix, obj.syntheticPrefix, i,
@@ -1034,8 +1013,8 @@ func (o *ingestExternalFilesOp) run(t *Test, h historyRecorder) {
 			err = db.Ingest(paths)
 		}
 	} else {
-		external := make([]pebble.ExternalFile, len(objs))
-		for i, obj := range objs {
+		external := make([]pebble.ExternalFile, len(o.objs))
+		for i, obj := range o.objs {
 			meta := t.getExternalObj(obj.externalObjID)
 			external[i] = pebble.ExternalFile{
 				Locator:           "external",

options.go

@@ -682,10 +682,6 @@ type Options struct {
 		// major version is at least `FormatFlushableIngest`.
 		DisableIngestAsFlushable func() bool
 
-		// CheckExternalIngestions enables opening external ssts at ingest time and
-		// validating that they are not empty. Used for testing/debugging.
-		CheckExternalIngestions bool
-
 		// RemoteStorage enables use of remote storage (e.g. S3) for storing
 		// sstables. Setting this option enables use of CreateOnShared option and
 		// allows ingestion of external files.

sstable/copier.go

@@ -31,7 +31,10 @@ import (
 //
 // The resulting sstable will have no block properties.
 //
-// Closes input and finishes or aborts output, including on non-nil errors.
+// The function might return ErrEmptySpan if there are no blocks that could
+// include keys in the given range. See ErrEmptySpan for more details.
+//
+// Closes input and finishes or aborts output in all cases, including on errors.
 //
 // Note that CopySpan is not aware of any suffix or prefix replacement; the
 // caller must account for those when specifying the bounds.
@@ -81,7 +84,8 @@ func CopySpan(
 	}()
 
 	if r.Properties.NumValueBlocks > 0 || r.Properties.NumRangeKeys() > 0 || r.Properties.NumRangeDeletions > 0 {
-		return 0, errors.New("cannot CopySpan sstables with value blocks or range keys")
+		// We just checked for these conditions above.
+		return 0, base.AssertionFailedf("cannot CopySpan sstables with value blocks or range keys")
 	}
 
 	// Set the filter block to be copied over if it exists. It will return false
@@ -128,7 +132,7 @@ func CopySpan(
 	// a non-empty sst by copying something outside the span, but #3907 means that
 	// the empty virtual span would still be a problem, so don't bother.
 	if len(blocks) < 1 {
-		return 0, errors.Newf("CopySpan cannot copy empty span %s %s", start, end)
+		return 0, ErrEmptySpan
 	}
 
 	// Find the span of the input file that contains all our blocks, and then copy
@@ -171,6 +175,15 @@ func CopySpan(
 	return wrote, nil
 }
 
+// ErrEmptySpan is returned by CopySpan if the input sstable has no keys in the
+// requested span.
+//
+// Note that CopySpan's determination of block overlap is best effort - we may
+// copy a block that doesn't actually contain any keys in the span, in which
+// case we won't generate this error. We currently only generate this error when
+// the span start is beyond all keys in the physical sstable.
+var ErrEmptySpan = errors.New("cannot copy empty span")
+
 // indexEntry captures the two components of an sst index entry: the key and the
 // decoded block handle value.
 type indexEntry struct {