manifest: add encoding of prefix rules to manifest entries

internal/manifest/version.go

@@ -269,6 +269,15 @@ type FileMetadata struct {
 	boundTypeSmallest, boundTypeLargest boundType
 	// Virtual is true if the FileMetadata belongs to a virtual sstable.
 	Virtual bool
+
+	// PrefixReplacement is used for virtual files where the backing file has a
+	// different prefix on its keys than the span in which it is being exposed.
+	PrefixReplacement *PrefixReplacement
+}
+
+// PrefixReplacement represents a read-time replacement of a key prefix.
+type PrefixReplacement struct {
+	ContentPrefix, SyntheticPrefix []byte
 }
 
 // PhysicalFileMeta is used by functions which want a guarantee that their input
@@ -851,6 +860,18 @@ func (m *FileMetadata) Validate(cmp Compare, formatKey base.FormatKey) error {
 		return base.CorruptionErrorf("file metadata FileBacking not set")
 	}
 
+	if m.PrefixReplacement != nil {
+		if !m.Virtual {
+			return base.CorruptionErrorf("prefix replacement rule set with non-virtual file")
+		}
+		if !bytes.HasPrefix(m.Smallest.UserKey, m.PrefixReplacement.SyntheticPrefix) {
+			return base.CorruptionErrorf("virtual file with prefix replacement rules has smallest key with a different prefix: %s", m.Smallest.Pretty(formatKey))
+		}
+		if !bytes.HasPrefix(m.Largest.UserKey, m.PrefixReplacement.SyntheticPrefix) {
+			return base.CorruptionErrorf("virtual file with prefix replacement rules has largest key with a different prefix: %s", m.Largest.Pretty(formatKey))
+		}
+	}
+
 	return nil
 }
 

internal/manifest/version_edit.go

@@ -63,6 +63,7 @@ const (
 	customTagPathID            = 65
 	customTagNonSafeIgnoreMask = 1 << 6
 	customTagVirtual           = 66
+	customTagPrefixRewrite     = 67
 )
 
 // DeletedFileEntry holds the state for a file deletion from a level. The file
@@ -336,6 +337,7 @@ func (v *VersionEdit) Decode(r io.Reader) error {
 				virtual        bool
 				backingFileNum uint64
 			}{}
+			var virtualPrefix *PrefixReplacement
 			if tag == tagNewFile4 || tag == tagNewFile5 {
 				for {
 					customTag, err := d.readUvarint()
@@ -352,6 +354,20 @@ func (v *VersionEdit) Decode(r io.Reader) error {
 						}
 						virtualState.backingFileNum = n
 						continue
+					} else if customTag == customTagPrefixRewrite {
+						content, err := d.readBytes()
+						if err != nil {
+							return err
+						}
+						synthetic, err := d.readBytes()
+						if err != nil {
+							return err
+						}
+						virtualPrefix = &PrefixReplacement{
+							ContentPrefix:   content,
+							SyntheticPrefix: synthetic,
+						}
+						continue
 					}
 
 					field, err := d.readBytes()
@@ -390,6 +406,7 @@ func (v *VersionEdit) Decode(r io.Reader) error {
 				LargestSeqNum:       largestSeqNum,
 				MarkedForCompaction: markedForCompaction,
 				Virtual:             virtualState.virtual,
+				PrefixReplacement:   virtualPrefix,
 			}
 			if tag != tagNewFile5 { // no range keys present
 				m.SmallestPointKey = base.DecodeInternalKey(smallestPointKey)
@@ -606,6 +623,11 @@ func (v *VersionEdit) Encode(w io.Writer) error {
 				e.writeUvarint(customTagVirtual)
 				e.writeUvarint(uint64(x.Meta.FileBacking.DiskFileNum.FileNum()))
 			}
+			if x.Meta.PrefixReplacement != nil {
+				e.writeUvarint(customTagPrefixRewrite)
+				e.writeBytes(x.Meta.PrefixReplacement.ContentPrefix)
+				e.writeBytes(x.Meta.PrefixReplacement.SyntheticPrefix)
+			}
 			e.writeUvarint(customTagTerminate)
 		}
 	}

sstable/prefix_replacing_iterator.go

@@ -159,7 +159,7 @@ func (p *prefixReplacingIterator) SetBounds(lower, upper []byte) {
 	// Check if the underlying iterator requires un-rewritten bounds, i.e. if it
 	// is going to rewrite them itself or pass them to something e.g. vState that
 	// will rewrite them.
-	if x, ok := p.i.(interface{ SetBoundsWithVirtualPrefix() bool }); ok && x.SetBoundsWithVirtualPrefix() {
+	if x, ok := p.i.(interface{ SetBoundsWithSyntheticPrefix() bool }); ok && x.SetBoundsWithSyntheticPrefix() {
 		p.i.SetBounds(lower, upper)
 		return
 	}
@@ -221,9 +221,6 @@ func (p *prefixReplacingFragmentIterator) rewriteSpan(sp *keyspan.Span) *keyspan
 	}
 	sp.Start = append(p.out1[:len(p.dst)], sp.Start[len(p.src):]...)
 	sp.End = append(p.out2[:len(p.dst)], sp.End[len(p.src):]...)
-
-	// TODO(dt): RESOLVE DURING CODE REVIEW
-	// do I need to touch sp.Keys? is sp.Start/End actually assured to both have dst prefix?
 	return sp
 }