Merge cockroachdb#41958

41958: storage/engine: make pebbleTimeBoundPropCollector extract intent timestamps r=itsbilal a=petermattis

Fixes cockroachdb#41900

Release note: None

Co-authored-by: Peter Mattis <petermattis@gmail.com>

pkg/storage/engine/pebble.go

@@ -88,10 +88,24 @@ var MVCCMerger = &pebble.Merger{
 }
 
 // pebbleTimeBoundPropCollector implements a property collector for MVCC
-// Timestamps.  Its behavior matches TimeBoundTblPropCollector in
+// Timestamps. Its behavior matches TimeBoundTblPropCollector in
 // table_props.cc.
+//
+// The handling of timestamps in intents is mildly complicated. Consider:
+//
+//   a@<meta>   -> <MVCCMetadata: Timestamp=t2>
+//   a@t2       -> <value>
+//   a@t1       -> <value>
+//
+// The metadata record (a.k.a. the intent) for a key always sorts first. The
+// timestamp field always points to the next record. In this case, the meta
+// record contains t2 and the next record is t2. Because of this duplication of
+// the timestamp both in the intent and in the timestamped record that
+// immediately follows it, we only need to unmarshal the MVCCMetadata if it is
+// the last key in the sstable.
 type pebbleTimeBoundPropCollector struct {
-	min, max []byte
+	min, max  []byte
+	lastValue []byte
 }
 
 func (t *pebbleTimeBoundPropCollector) Add(key pebble.InternalKey, value []byte) error {
@@ -100,22 +114,45 @@ func (t *pebbleTimeBoundPropCollector) Add(key pebble.InternalKey, value []byte)
 		return errors.Errorf("failed to split MVCC key")
 	}
 	if len(ts) > 0 {
-		if len(t.min) == 0 || bytes.Compare(ts, t.min) < 0 {
-			t.min = append(t.min[:0], ts...)
-		}
-		if len(t.max) == 0 || bytes.Compare(ts, t.max) > 0 {
-			t.max = append(t.max[:0], ts...)
-		}
+		t.lastValue = t.lastValue[:0]
+		t.updateBounds(ts)
+	} else {
+		t.lastValue = append(t.lastValue[:0], value...)
 	}
 	return nil
 }
 
 func (t *pebbleTimeBoundPropCollector) Finish(userProps map[string]string) error {
+	if len(t.lastValue) > 0 {
+		// The last record in the sstable was an intent. Unmarshal the metadata and
+		// update the bounds with the timestamp it contains.
+		meta := &enginepb.MVCCMetadata{}
+		if err := protoutil.Unmarshal(t.lastValue, meta); err != nil {
+			// We're unable to parse the MVCCMetadata. Fail open by not setting the
+			// min/max timestamp properties. THis mimics the behavior of
+			// TimeBoundTblPropCollector.
+			return nil
+		}
+		if meta.Txn != nil {
+			ts := encodeTimestamp(hlc.Timestamp(meta.Timestamp))
+			t.updateBounds(ts)
+		}
+	}
+
 	userProps["crdb.ts.min"] = string(t.min)
 	userProps["crdb.ts.max"] = string(t.max)
 	return nil
 }
 
+func (t *pebbleTimeBoundPropCollector) updateBounds(ts []byte) {
+	if len(t.min) == 0 || bytes.Compare(ts, t.min) < 0 {
+		t.min = append(t.min[:0], ts...)
+	}
+	if len(t.max) == 0 || bytes.Compare(ts, t.max) > 0 {
+		t.max = append(t.max[:0], ts...)
+	}
+}
+
 func (t *pebbleTimeBoundPropCollector) Name() string {
 	// This constant needs to match the one used by the RocksDB version of this
 	// table property collector. DO NOT CHANGE.

pkg/storage/engine/pebble_test.go

@@ -0,0 +1,101 @@
+// Copyright 2019 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package engine
+
+import (
+	"bytes"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble"
+)
+
+func TestPebbleTimeBoundPropCollector(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	datadriven.RunTest(t, "testdata/time_bound_props", func(d *datadriven.TestData) string {
+		c := &pebbleTimeBoundPropCollector{}
+		switch d.Cmd {
+		case "build":
+			for _, line := range strings.Split(d.Input, "\n") {
+				parts := strings.Fields(line)
+				if len(parts) != 2 {
+					return fmt.Sprintf("malformed line: %s, expected: <key>/<timestamp> <value>", line)
+				}
+				keyParts := strings.Split(parts[0], "/")
+				if len(keyParts) != 2 {
+					return fmt.Sprintf("malformed key: %s, expected: <key>/<timestamp>", parts[0])
+				}
+
+				key := []byte(keyParts[0])
+				timestamp, err := strconv.Atoi(keyParts[1])
+				if err != nil {
+					return err.Error()
+				}
+				ikey := pebble.InternalKey{
+					UserKey: EncodeKey(MVCCKey{
+						Key:       key,
+						Timestamp: hlc.Timestamp{WallTime: int64(timestamp)},
+					}),
+				}
+
+				value := []byte(parts[1])
+				if timestamp == 0 {
+					if n, err := fmt.Sscanf(string(value), "timestamp=%d", &timestamp); err != nil {
+						return err.Error()
+					} else if n != 1 {
+						return fmt.Sprintf("malformed txn timestamp: %s, expected timestamp=<value>", value)
+					}
+					meta := &enginepb.MVCCMetadata{}
+					meta.Timestamp.WallTime = int64(timestamp)
+					meta.Txn = &enginepb.TxnMeta{}
+					var err error
+					value, err = protoutil.Marshal(meta)
+					if err != nil {
+						return err.Error()
+					}
+				}
+
+				if err := c.Add(ikey, value); err != nil {
+					return err.Error()
+				}
+			}
+
+			// Retrieve the properties and sort them for test determinism.
+			m := make(map[string]string)
+			if err := c.Finish(m); err != nil {
+				return err.Error()
+			}
+			var keys []string
+			for k := range m {
+				keys = append(keys, k)
+			}
+			sort.Strings(keys)
+
+			var buf bytes.Buffer
+			for _, k := range keys {
+				fmt.Fprintf(&buf, "%s: %x\n", k, m[k])
+			}
+			return buf.String()
+
+		default:
+			return fmt.Sprintf("unknown command: %s", d.Cmd)
+		}
+	})
+}

pkg/storage/engine/testdata/time_bound_props

@@ -0,0 +1,46 @@
+# build
+# key/0 timestamp=value
+# key/timestamp value
+# ...
+# ----
+
+build
+a/2 b
+a/1 c
+----
+crdb.ts.max: 0000000000000002
+crdb.ts.min: 0000000000000001
+
+build
+a/0 timestamp=2
+a/2 b
+a/1 c
+----
+crdb.ts.max: 0000000000000002
+crdb.ts.min: 0000000000000001
+
+build
+a/0 timestamp=2
+----
+crdb.ts.max: 0000000000000002
+crdb.ts.min: 0000000000000002
+
+# Note that this case should never happen. We can't have an intent on
+# "a" without a corresponding timestamped value.
+
+build
+a/0 timestamp=2
+b/0 timestamp=3
+----
+crdb.ts.max: 0000000000000003
+crdb.ts.min: 0000000000000003
+
+# Note that this case should never happen. The timestamp in the intent
+# on "a" should match the subsequent timestamped value.
+
+build
+a/0 timestamp=3
+a/1 b
+----
+crdb.ts.max: 0000000000000001
+crdb.ts.min: 0000000000000001