storage: add MVCCTimeInterval block property for range keys

This patch adds `MVCCTimeInternal` block property collection and
filtering for range keys, which allows using time-bound iterators with
range keys.

Range keys will only be written once the `MVCCRangeTombstones` version
gate is enabled.

Release note: None

pkg/storage/pebble.go

@@ -325,25 +325,66 @@ var MVCCMerger = &pebble.Merger{
 	},
 }
 
-// pebbleDataBlockMVCCTimeIntervalCollector provides an implementation of
+// pebbleDataBlockMVCCTimeIntervalPointCollector implements
+// pebble.DataBlockIntervalCollector for point keys.
+type pebbleDataBlockMVCCTimeIntervalPointCollector struct {
+	pebbleDataBlockMVCCTimeIntervalCollector
+}
+
+var (
+	_ sstable.DataBlockIntervalCollector      = (*pebbleDataBlockMVCCTimeIntervalPointCollector)(nil)
+	_ sstable.SuffixReplaceableBlockCollector = (*pebbleDataBlockMVCCTimeIntervalPointCollector)(nil)
+)
+
+func (tc *pebbleDataBlockMVCCTimeIntervalPointCollector) Add(
+	key pebble.InternalKey, _ []byte,
+) error {
+	return tc.add(key.UserKey)
+}
+
+// pebbleDataBlockMVCCTimeIntervalRangeCollector implements
+// pebble.DataBlockIntervalCollector for range keys.
+type pebbleDataBlockMVCCTimeIntervalRangeCollector struct {
+	pebbleDataBlockMVCCTimeIntervalCollector
+}
+
+var (
+	_ sstable.DataBlockIntervalCollector      = (*pebbleDataBlockMVCCTimeIntervalRangeCollector)(nil)
+	_ sstable.SuffixReplaceableBlockCollector = (*pebbleDataBlockMVCCTimeIntervalRangeCollector)(nil)
+)
+
+func (tc *pebbleDataBlockMVCCTimeIntervalRangeCollector) Add(
+	key pebble.InternalKey, value []byte,
+) error {
+	// TODO(erikgrinaker): should reuse a buffer for keysDst, but keyspan.Key is
+	// not exported by Pebble.
+	span, err := sstable.DecodeRangeKey(key, value, nil)
+	if err != nil {
+		return errors.Wrapf(err, "decoding range key at %s", key)
+	}
+	for _, k := range span.Keys {
+		if err := tc.add(k.Suffix); err != nil {
+			return errors.Wrapf(err, "recording suffix %x for range key at %s", k.Suffix, key)
+		}
+	}
+	return nil
+}
+
+// pebbleDataBlockMVCCTimeIntervalCollector is a helper for a
 // pebble.DataBlockIntervalCollector that is used to construct a
 // pebble.BlockPropertyCollector. This provides per-block filtering, which
 // also gets aggregated to the sstable-level and filters out sstables. It must
 // only be used for MVCCKeyIterKind iterators, since it will ignore
 // blocks/sstables that contain intents (and any other key that is not a real
 // MVCC key).
+//
+// This is wrapped by structs for point or range key collection, which actually
+// implement pebble.DataBlockIntervalCollector.
 type pebbleDataBlockMVCCTimeIntervalCollector struct {
 	// min, max are the encoded timestamps.
 	min, max []byte
 }
 
-var _ sstable.DataBlockIntervalCollector = &pebbleDataBlockMVCCTimeIntervalCollector{}
-var _ sstable.SuffixReplaceableBlockCollector = (*pebbleDataBlockMVCCTimeIntervalCollector)(nil)
-
-func (tc *pebbleDataBlockMVCCTimeIntervalCollector) Add(key pebble.InternalKey, _ []byte) error {
-	return tc.add(key.UserKey)
-}
-
 // add collects the given slice in the collector. The slice may be an entire
 // encoded MVCC key, or the bare suffix of an encoded key.
 func (tc *pebbleDataBlockMVCCTimeIntervalCollector) add(b []byte) error {
@@ -431,8 +472,8 @@ var PebbleBlockPropertyCollectors = []func() pebble.BlockPropertyCollector{
 	func() pebble.BlockPropertyCollector {
 		return sstable.NewBlockIntervalCollector(
 			mvccWallTimeIntervalCollector,
-			&pebbleDataBlockMVCCTimeIntervalCollector{}, /* points */
-			nil, /* ranges */
+			&pebbleDataBlockMVCCTimeIntervalPointCollector{},
+			&pebbleDataBlockMVCCTimeIntervalRangeCollector{},
 		)
 	},
 }

pkg/storage/pebble_iterator.go

@@ -239,6 +239,11 @@ func (p *pebbleIterator) setOptions(opts IterOptions, durability DurabilityRequi
 				uint64(opts.MinTimestampHint.WallTime),
 				uint64(opts.MaxTimestampHint.WallTime)+1),
 		}
+		p.options.RangeKeyFilters = []pebble.BlockPropertyFilter{
+			sstable.NewBlockIntervalFilter(mvccWallTimeIntervalCollector,
+				uint64(opts.MinTimestampHint.WallTime),
+				uint64(opts.MaxTimestampHint.WallTime)+1),
+		}
 	}
 
 	// Set the new iterator options. We unconditionally do so, since Pebble will