sstable: refactor prop block load

We now pass in an iterator to `load` in order to range all over the
keys and values.

Author: annrpom

sstable/colblk/key_value_block.go

@@ -5,6 +5,8 @@
 package colblk
 
 import (
+	"iter"
+
 	"github.com/cockroachdb/pebble/internal/binfmt"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 )
@@ -124,3 +126,14 @@ func (r *KeyValueBlockDecoder) KeyAt(i int) []byte {
 func (r *KeyValueBlockDecoder) ValueAt(i int) []byte {
 	return r.values.At(i)
 }
+
+// All returns an iterator that ranges over all key-value pairs in the block.
+func (r *KeyValueBlockDecoder) All() iter.Seq2[[]byte, []byte] {
+	return func(yield func([]byte, []byte) bool) {
+		for i := 0; i < r.BlockDecoder().Rows(); i++ {
+			if !yield(r.KeyAt(i), r.ValueAt(i)) {
+				return
+			}
+		}
+	}
+}

sstable/layout.go

@@ -574,11 +574,17 @@ func decodeLayout(comparer *base.Comparer, data []byte, tableFormat TableFormat)
 		return Layout{}, errors.Wrap(err, "decompressing properties")
 	}
 	if tableFormat >= TableFormatPebblev6 {
-		if err = props.load(decompressedProps, map[string]struct{}{}); err != nil {
+		var decoder colblk.KeyValueBlockDecoder
+		decoder.Init(decompressedProps)
+		if err = props.load(decoder.All(), map[string]struct{}{}); err != nil {
 			return Layout{}, err
 		}
 	} else {
-		if err = props.loadFromRowBlock(decompressedProps, map[string]struct{}{}); err != nil {
+		i, err := rowblk.NewRawIter(bytes.Compare, decompressedProps)
+		if err != nil {
+			return Layout{}, err
+		}
+		if err = props.load(i.All(), map[string]struct{}{}); err != nil {
 			return Layout{}, err
 		}
 	}

sstable/properties.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"iter"
 	"maps"
 	"math"
 	"reflect"
@@ -274,66 +275,26 @@ func (p *Properties) String() string {
 	return buf.String()
 }
 
-func (p *Properties) loadFromRowBlock(b []byte, deniedUserProperties map[string]struct{}) error {
-	i, err := rowblk.NewRawIter(bytes.Compare, b)
-	if err != nil {
-		return err
-	}
-	p.Loaded = make(map[uintptr]struct{})
-	v := reflect.ValueOf(p).Elem()
-
-	for valid := i.First(); valid; valid = i.Next() {
-		if f, ok := propTagMap[string(i.Key().UserKey)]; ok {
-			p.Loaded[f.Offset] = struct{}{}
-			field := v.FieldByIndex(f.Index)
-			switch f.Type.Kind() {
-			case reflect.Bool:
-				field.SetBool(bytes.Equal(i.Value(), propBoolTrue))
-			case reflect.Uint32:
-				field.SetUint(uint64(binary.LittleEndian.Uint32(i.Value())))
-			case reflect.Uint64:
-				n, _ := binary.Uvarint(i.Value())
-				field.SetUint(n)
-			case reflect.String:
-				field.SetString(intern.Bytes(i.Value()))
-			default:
-				panic("not reached")
-			}
-			continue
-		}
-		if p.UserProperties == nil {
-			p.UserProperties = make(map[string]string)
-		}
-
-		if _, denied := deniedUserProperties[string(i.Key().UserKey)]; !denied {
-			p.UserProperties[intern.Bytes(i.Key().UserKey)] = string(i.Value())
-		}
-	}
-	return nil
-}
-
-func (p *Properties) load(b []byte, deniedUserProperties map[string]struct{}) error {
-	var decoder colblk.KeyValueBlockDecoder
-	decoder.Init(b)
+func (p *Properties) load(
+	i iter.Seq2[[]byte, []byte], deniedUserProperties map[string]struct{},
+) error {
 	p.Loaded = make(map[uintptr]struct{})
 	v := reflect.ValueOf(p).Elem()
 
-	for i := 0; i < decoder.BlockDecoder().Rows(); i++ {
-		key := decoder.KeyAt(i)
-		value := decoder.ValueAt(i)
+	for key, val := range i {
 		if f, ok := propTagMap[string(key)]; ok {
 			p.Loaded[f.Offset] = struct{}{}
 			field := v.FieldByIndex(f.Index)
 			switch f.Type.Kind() {
 			case reflect.Bool:
-				field.SetBool(bytes.Equal(value, propBoolTrue))
+				field.SetBool(bytes.Equal(val, propBoolTrue))
 			case reflect.Uint32:
-				field.SetUint(uint64(binary.LittleEndian.Uint32(value)))
+				field.SetUint(uint64(binary.LittleEndian.Uint32(val)))
 			case reflect.Uint64:
-				n, _ := binary.Uvarint(value)
+				n, _ := binary.Uvarint(val)
 				field.SetUint(n)
 			case reflect.String:
-				field.SetString(intern.Bytes(value))
+				field.SetString(intern.Bytes(val))
 			default:
 				panic("not reached")
 			}
@@ -344,7 +305,7 @@ func (p *Properties) load(b []byte, deniedUserProperties map[string]struct{}) er
 		}
 
 		if _, denied := deniedUserProperties[string(key)]; !denied {
-			p.UserProperties[intern.Bytes(key)] = string(value)
+			p.UserProperties[intern.Bytes(key)] = string(val)
 		}
 	}
 	return nil

sstable/properties_test.go

@@ -5,6 +5,7 @@
 package sstable
 
 import (
+	"bytes"
 	randv1 "math/rand"
 	"path/filepath"
 	"reflect"
@@ -106,7 +107,9 @@ func TestPropertiesSave(t *testing.T) {
 		require.NoError(t, e.saveToRowWriter(TableFormatPebblev2, &w))
 		var props Properties
 
-		require.NoError(t, props.loadFromRowBlock(w.Finish(), make(map[string]struct{})))
+		i, err := rowblk.NewRawIter(bytes.Compare, w.Finish())
+		require.NoError(t, err)
+		require.NoError(t, props.load(i.All(), make(map[string]struct{})))
 		props.Loaded = nil
 		if diff := pretty.Diff(*e, props); diff != nil {
 			t.Fatalf("%s", strings.Join(diff, "\n"))
@@ -137,6 +140,8 @@ func BenchmarkPropertiesLoad(b *testing.B) {
 	p := &Properties{}
 	for i := 0; i < b.N; i++ {
 		*p = Properties{}
-		require.NoError(b, p.loadFromRowBlock(block, nil))
+		it, err := rowblk.NewRawIter(bytes.Compare, block)
+		require.NoError(b, err)
+		require.NoError(b, p.load(it.All(), nil))
 	}
 }

sstable/reader.go

@@ -5,6 +5,7 @@
 package sstable
 
 import (
+	"bytes"
 	"cmp"
 	"context"
 	"fmt"
@@ -446,14 +447,21 @@ func (r *Reader) readMetaindex(
 		}
 		r.propertiesBH = bh
 		if r.tableFormat >= TableFormatPebblev6 {
-			err = r.Properties.load(b.BlockData(), deniedUserProperties)
+			var decoder colblk.KeyValueBlockDecoder
+			decoder.Init(b.BlockData())
+			if err := r.Properties.load(decoder.All(), deniedUserProperties); err != nil {
+				return err
+			}
 		} else {
-			err = r.Properties.loadFromRowBlock(b.BlockData(), deniedUserProperties)
+			i, err := rowblk.NewRawIter(bytes.Compare, b.BlockData())
+			if err != nil {
+				return err
+			}
+			if err := r.Properties.load(i.All(), deniedUserProperties); err != nil {
+				return err
+			}
 		}
 		b.Release()
-		if err != nil {
-			return err
-		}
 	} else {
 		return errors.New("did not read any value for the properties block in the meta index")
 	}

sstable/rowblk/rowblk_iter.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"encoding/binary"
 	"io"
+	"iter"
 	"slices"
 	"sort"
 	"unsafe"
@@ -1943,6 +1944,17 @@ func (i *RawIter) Describe(tp treeprinter.Node, fmtKV DescribeKV) {
 	}
 }
 
+// All returns an iterator that ranges over all key-value pairs in the block.
+func (i *RawIter) All() iter.Seq2[[]byte, []byte] {
+	return func(yield func([]byte, []byte) bool) {
+		for valid := i.First(); valid; valid = i.Next() {
+			if !yield(i.Key().UserKey, i.Value()) {
+				return
+			}
+		}
+	}
+}
+
 func decodeVarint(ptr unsafe.Pointer) (uint32, unsafe.Pointer) {
 	if a := *((*uint8)(ptr)); a < 128 {
 		return uint32(a),