colblk: use key value columnar block for properties

This patch uses the new, drop-in columnar block to
replace the row format of properties blocks.

Fixes: #4425

Author: annrpom

format_major_version.go

@@ -208,8 +208,9 @@ const (
 	// format TableFormatPebblev6.
 	//
 	// The TableFormatPebblev6 sstable format introduces a checksum within the
-	// sstable footer, and allows inclusion of blob handle references within the
-	// value column of a sstable block.
+	// sstable footer, allows inclusion of blob handle references within the
+	// value column of a sstable block, and supports columnar meta index +
+	// properties blocks.
 	//
 	// This format major version does not yet enable use of value separation.
 	FormatTableFormatV6

sstable/colblk/key_value_block_test.go

@@ -48,3 +48,29 @@ func TestMetaIndexBlock(t *testing.T) {
 		}
 	})
 }
+
+func TestPropertiesBlock(t *testing.T) {
+	var decoder KeyValueBlockDecoder
+	var buf bytes.Buffer
+	datadriven.RunTest(t, "testdata/properties_block", func(t *testing.T, d *datadriven.TestData) string {
+		buf.Reset()
+		switch d.Cmd {
+		case "build":
+			var w KeyValueBlockWriter
+			w.Init()
+			for _, line := range strings.Split(d.Input, "\n") {
+				fields := strings.Fields(line)
+				key := []byte(fields[0])
+				value := []byte(fields[1])
+				w.AddKV(key, value)
+			}
+
+			data := w.Finish(w.Rows())
+			decoder.Init(data)
+			fmt.Fprint(&buf, decoder.DebugString())
+			return buf.String()
+		default:
+			return fmt.Sprintf("unknown command: %s", d.Cmd)
+		}
+	})
+}

sstable/colblk/testdata/properties_block

@@ -0,0 +1,48 @@
+build
+rocksdb.deleted.keys       14
+rocksdb.filter.size        2
+rocksdb.index.size         3
+rocksdb.num.data.blocks    8
+----
+key-value-block-decoder
+ └── key value block header
+      ├── columnar block header
+      │    ├── 000-001: x 01       # version 1
+      │    ├── 001-003: x 0200     # 2 columns
+      │    ├── 003-007: x 04000000 # 4 rows
+      │    ├── 007-008: b 00000011 # col 0: bytes
+      │    ├── 008-012: x 11000000 # col 0: page start 17
+      │    ├── 012-013: b 00000011 # col 1: bytes
+      │    └── 013-017: x 67000000 # col 1: page start 103
+      ├── data for column 0 (bytes)
+      │    ├── offsets table
+      │    │    ├── 017-018: x 01 # encoding: 1b
+      │    │    ├── 018-019: x 00 # data[0] = 0 [23 overall]
+      │    │    ├── 019-020: x 14 # data[1] = 20 [43 overall]
+      │    │    ├── 020-021: x 27 # data[2] = 39 [62 overall]
+      │    │    ├── 021-022: x 39 # data[3] = 57 [80 overall]
+      │    │    └── 022-023: x 50 # data[4] = 80 [103 overall]
+      │    └── data
+      │         ├── 023-033: x 726f636b7364622e6465 # data[0]: rocksdb.deleted.keys
+      │         ├── 033-043: x 6c657465642e6b657973 # (continued...)
+      │         ├── 043-053: x 726f636b7364622e6669 # data[1]: rocksdb.filter.size
+      │         ├── 053-062: x 6c7465722e73697a65   # (continued...)
+      │         ├── 062-072: x 726f636b7364622e696e # data[2]: rocksdb.index.size
+      │         ├── 072-080: x 6465782e73697a65     # (continued...)
+      │         ├── 080-090: x 726f636b7364622e6e75 # data[3]: rocksdb.num.data.blocks
+      │         ├── 090-100: x 6d2e646174612e626c6f # (continued...)
+      │         └── 100-103: x 636b73               # (continued...)
+      ├── data for column 1 (bytes)
+      │    ├── offsets table
+      │    │    ├── 103-104: x 01 # encoding: 1b
+      │    │    ├── 104-105: x 00 # data[0] = 0 [109 overall]
+      │    │    ├── 105-106: x 02 # data[1] = 2 [111 overall]
+      │    │    ├── 106-107: x 03 # data[2] = 3 [112 overall]
+      │    │    ├── 107-108: x 04 # data[3] = 4 [113 overall]
+      │    │    └── 108-109: x 05 # data[4] = 5 [114 overall]
+      │    └── data
+      │         ├── 109-111: x 3134 # data[0]: 14
+      │         ├── 111-112: x 32   # data[1]: 2
+      │         ├── 112-113: x 33   # data[2]: 3
+      │         └── 113-114: x 38   # data[3]: 8
+      └── 114-115: x 00 # block padding byte

sstable/colblk_writer.go

@@ -1011,16 +1011,25 @@ func (w *RawColumnWriter) Close() (err error) {
 			}
 		}
 
-		var raw rowblk.Writer
-		// The restart interval is set to infinity because the properties block
-		// is always read sequentially and cached in a heap located object. This
-		// reduces table size without a significant impact on performance.
-		raw.RestartInterval = propertiesBlockRestartInterval
+		var toWrite []byte
 		w.props.CompressionOptions = rocksDBCompressionOptions
-		if err := w.props.save(w.opts.TableFormat, &raw); err != nil {
-			return err
+		if w.opts.TableFormat >= TableFormatPebblev6 {
+			var cw colblk.KeyValueBlockWriter
+			cw.Init()
+			w.props.saveToColWriter(w.opts.TableFormat, &cw)
+			toWrite = cw.Finish(cw.Rows())
+		} else {
+			var raw rowblk.Writer
+			// The restart interval is set to infinity because the properties block
+			// is always read sequentially and cached in a heap located object. This
+			// reduces table size without a significant impact on performance.
+			raw.RestartInterval = propertiesBlockRestartInterval
+			if err = w.props.saveToRowWriter(w.opts.TableFormat, &raw); err != nil {
+				return err
+			}
+			toWrite = raw.Finish()
 		}
-		if _, err := w.layout.WritePropertiesBlock(raw.Finish()); err != nil {
+		if _, err = w.layout.WritePropertiesBlock(toWrite); err != nil {
 			return err
 		}
 	}

sstable/format.go

@@ -29,7 +29,7 @@ const (
 	TableFormatPebblev3 // Value blocks.
 	TableFormatPebblev4 // DELSIZED tombstones.
 	TableFormatPebblev5 // Columnar blocks.
-	TableFormatPebblev6 // Checksum footer + blob value handles + columnar metaindex.
+	TableFormatPebblev6 // Checksum footer + blob value handles + columnar metaindex/properties.
 	NumTableFormats
 
 	TableFormatMax = NumTableFormats - 1

sstable/layout.go

@@ -254,10 +254,23 @@ func (l *Layout) Describe(
 				if err != nil {
 					return err
 				}
-				iter, _ := rowblk.NewRawIter(r.Comparer.Compare, h.BlockData())
-				iter.Describe(tpNode, func(w io.Writer, key *base.InternalKey, value []byte, enc rowblk.KVEncoding) {
-					fmt.Fprintf(w, "%05d    %s (%d)", enc.Offset, key.UserKey, enc.Length)
-				})
+				if r.tableFormat >= TableFormatPebblev6 {
+					var decoder colblk.KeyValueBlockDecoder
+					decoder.Init(h.BlockData())
+					offset := 0
+					for i := 0; i < decoder.BlockDecoder().Rows(); i++ {
+						key := decoder.KeyAt(i)
+						value := decoder.ValueAt(i)
+						length := len(key) + len(value)
+						tpNode.Childf("%05d    %s (%d)", offset, key, length)
+						offset += length
+					}
+				} else {
+					iter, _ := rowblk.NewRawIter(r.Comparer.Compare, h.BlockData())
+					iter.Describe(tpNode, func(w io.Writer, key *base.InternalKey, value []byte, enc rowblk.KVEncoding) {
+						fmt.Fprintf(w, "%05d    %s (%d)", enc.Offset, key.UserKey, enc.Length)
+					})
+				}
 
 			case "meta-index":
 				if b.Handle != r.metaindexBH {
@@ -286,18 +299,16 @@ func (l *Layout) Describe(
 							bh, n = block.DecodeHandle(value)
 						}
 						if n == 0 || n != len(value) {
-							s := fmt.Sprintf("%04d    [err: %s]\n", i, err)
-							tpNode.Child(s)
+							tpNode.Childf("%04d    [err: %s]\n", i, err)
 							continue
 						}
 						var vbihStr string
 						if isValueBlocksIndexHandle {
 							vbihStr = fmt.Sprintf(" value-blocks-index-lengths: %d(num), %d(offset), %d(length)",
 								vbih.BlockNumByteLength, vbih.BlockOffsetByteLength, vbih.BlockLengthByteLength)
 						}
-						s := fmt.Sprintf("%04d    %s block:%d/%d%s\n",
+						tpNode.Childf("%04d    %s block:%d/%d%s\n",
 							i, key, bh.Offset, bh.Length, vbihStr)
-						tpNode.Child(s)
 					}
 				} else {
 					iter, _ := rowblk.NewRawIter(r.Comparer.Compare, h.BlockData())
@@ -562,8 +573,14 @@ func decodeLayout(comparer *base.Comparer, data []byte, tableFormat TableFormat)
 	if err != nil {
 		return Layout{}, errors.Wrap(err, "decompressing properties")
 	}
-	if err := props.load(decompressedProps, map[string]struct{}{}); err != nil {
-		return Layout{}, err
+	if tableFormat >= TableFormatPebblev6 {
+		if err = props.load(decompressedProps, map[string]struct{}{}); err != nil {
+			return Layout{}, err
+		}
+	} else {
+		if err = props.loadFromRowBlock(decompressedProps, map[string]struct{}{}); err != nil {
+			return Layout{}, err
+		}
 	}
 
 	if props.IndexType == twoLevelIndex {
@@ -816,33 +833,38 @@ func (w *layoutWriter) WriteFilterBlock(f filterWriter) (bh block.Handle, err er
 	if err != nil {
 		return block.Handle{}, err
 	}
-	return w.writeNamedBlock(b, f.metaName())
+	return w.writeNamedBlock(b, block.NoCompression, f.metaName())
 }
 
 // WritePropertiesBlock constructs a trailer for the provided properties block
 // and writes the block and trailer to the writer. It automatically adds the
 // properties block to the file's meta index when the writer is finished.
 func (w *layoutWriter) WritePropertiesBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, metaPropertiesName)
+	if w.tableFormat >= TableFormatPebblev6 {
+		return w.writeNamedBlock(b, w.compression, metaPropertiesName)
+	}
+	return w.writeNamedBlock(b, block.NoCompression, metaPropertiesName)
 }
 
 // WriteRangeKeyBlock constructs a trailer for the provided range key block and
 // writes the block and trailer to the writer. It automatically adds the range
 // key block to the file's meta index when the writer is finished.
 func (w *layoutWriter) WriteRangeKeyBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, metaRangeKeyName)
+	return w.writeNamedBlock(b, block.NoCompression, metaRangeKeyName)
 }
 
 // WriteRangeDeletionBlock constructs a trailer for the provided range deletion
 // block and writes the block and trailer to the writer. It automatically adds
 // the range deletion block to the file's meta index when the writer is
 // finished.
 func (w *layoutWriter) WriteRangeDeletionBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, metaRangeDelV2Name)
+	return w.writeNamedBlock(b, block.NoCompression, metaRangeDelV2Name)
 }
 
-func (w *layoutWriter) writeNamedBlock(b []byte, name string) (bh block.Handle, err error) {
-	bh, err = w.writeBlock(b, block.NoCompression, &w.buf)
+func (w *layoutWriter) writeNamedBlock(
+	b []byte, compression block.Compression, name string,
+) (bh block.Handle, err error) {
+	bh, err = w.writeBlock(b, compression, &w.buf)
 	if err == nil {
 		w.recordToMetaindex(name, bh)
 	}

sstable/properties.go

@@ -8,13 +8,16 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"maps"
 	"math"
 	"reflect"
+	"slices"
 	"sort"
 	"strings"
 	"unsafe"
 
 	"github.com/cockroachdb/pebble/internal/intern"
+	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/sstable/rowblk"
 )
 
@@ -271,7 +274,7 @@ func (p *Properties) String() string {
 	return buf.String()
 }
 
-func (p *Properties) load(b []byte, deniedUserProperties map[string]struct{}) error {
+func (p *Properties) loadFromRowBlock(b []byte, deniedUserProperties map[string]struct{}) error {
 	i, err := rowblk.NewRawIter(bytes.Compare, b)
 	if err != nil {
 		return err
@@ -309,6 +312,44 @@ func (p *Properties) load(b []byte, deniedUserProperties map[string]struct{}) er
 	return nil
 }
 
+func (p *Properties) load(b []byte, deniedUserProperties map[string]struct{}) error {
+	var decoder colblk.KeyValueBlockDecoder
+	decoder.Init(b)
+	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)
+		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))
+			case reflect.Uint32:
+				field.SetUint(uint64(binary.LittleEndian.Uint32(value)))
+			case reflect.Uint64:
+				n, _ := binary.Uvarint(value)
+				field.SetUint(n)
+			case reflect.String:
+				field.SetString(intern.Bytes(value))
+			default:
+				panic("not reached")
+			}
+			continue
+		}
+		if p.UserProperties == nil {
+			p.UserProperties = make(map[string]string)
+		}
+
+		if _, denied := deniedUserProperties[string(key)]; !denied {
+			p.UserProperties[intern.Bytes(key)] = string(value)
+		}
+	}
+	return nil
+}
+
 func (p *Properties) saveBool(m map[string][]byte, offset uintptr, value bool) {
 	tag := propOffsetTagMap[offset]
 	if value {
@@ -342,7 +383,7 @@ func (p *Properties) saveString(m map[string][]byte, offset uintptr, value strin
 	m[propOffsetTagMap[offset]] = []byte(value)
 }
 
-func (p *Properties) save(tblFormat TableFormat, w *rowblk.Writer) error {
+func (p *Properties) accumulateProps(tblFormat TableFormat) ([]string, map[string][]byte) {
 	m := make(map[string][]byte)
 	for k, v := range p.UserProperties {
 		m[k] = []byte(v)
@@ -438,11 +479,14 @@ func (p *Properties) save(tblFormat TableFormat, w *rowblk.Writer) error {
 		m["rocksdb.format.version"] = singleZeroSlice
 	}
 
-	keys := make([]string, 0, len(m))
-	for key := range m {
-		keys = append(keys, key)
-	}
+	keys := slices.Collect(maps.Keys(m))
 	sort.Strings(keys)
+
+	return keys, m
+}
+
+func (p *Properties) saveToRowWriter(tblFormat TableFormat, w *rowblk.Writer) error {
+	keys, m := p.accumulateProps(tblFormat)
 	for _, key := range keys {
 		if err := w.AddRawString(key, m[key]); err != nil {
 			return err
@@ -451,6 +495,22 @@ func (p *Properties) save(tblFormat TableFormat, w *rowblk.Writer) error {
 	return nil
 }
 
+func (p *Properties) saveToColWriter(tblFormat TableFormat, w *colblk.KeyValueBlockWriter) {
+	keys, m := p.accumulateProps(tblFormat)
+	for _, key := range keys {
+		// Zero-length keys are unsupported. See below about StringData.
+		if len(key) == 0 {
+			continue
+		}
+		// Use an unsafe conversion to avoid allocating. AddKV is not
+		// supposed to modify the given slice, so the unsafe conversion
+		// is okay. Note that unsafe.StringData panics if len(key) == 0,
+		// so we explicitly skip zero-length keys above. They shouldn't
+		// occur in practice.
+		w.AddKV(unsafe.Slice(unsafe.StringData(key), len(key)), m[key])
+	}
+}
+
 var (
 	singleZeroSlice = []byte{0x00}
 	maxInt32Slice   = binary.AppendUvarint([]byte(nil), math.MaxInt32)

sstable/properties_test.go

@@ -103,10 +103,10 @@ func TestPropertiesSave(t *testing.T) {
 		// Check that we can save properties and read them back.
 		var w rowblk.Writer
 		w.RestartInterval = propertiesBlockRestartInterval
-		require.NoError(t, e.save(TableFormatPebblev2, &w))
+		require.NoError(t, e.saveToRowWriter(TableFormatPebblev2, &w))
 		var props Properties
 
-		require.NoError(t, props.load(w.Finish(), make(map[string]struct{})))
+		require.NoError(t, props.loadFromRowBlock(w.Finish(), make(map[string]struct{})))
 		props.Loaded = nil
 		if diff := pretty.Diff(*e, props); diff != nil {
 			t.Fatalf("%s", strings.Join(diff, "\n"))
@@ -130,13 +130,13 @@ func TestPropertiesSave(t *testing.T) {
 func BenchmarkPropertiesLoad(b *testing.B) {
 	var w rowblk.Writer
 	w.RestartInterval = propertiesBlockRestartInterval
-	require.NoError(b, testProps.save(TableFormatPebblev2, &w))
+	require.NoError(b, testProps.saveToRowWriter(TableFormatPebblev2, &w))
 	block := w.Finish()
 
 	b.ResetTimer()
 	p := &Properties{}
 	for i := 0; i < b.N; i++ {
 		*p = Properties{}
-		require.NoError(b, p.load(block, nil))
+		require.NoError(b, p.loadFromRowBlock(block, nil))
 	}
 }