kv: maximally compress persisted prior read summaries

These are rarely read, so it is not worth the extra space to store them
with high precision. Instead, maximally compress them before persisting
them.

Epic: None
Release note: None

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -1305,6 +1305,8 @@ func mergeTrigger(
 		} else if priorSum != nil {
 			mergedSum.Merge(*priorSum)
 		}
+		// Compress the persisted read summary, as it will likely never be needed.
+		mergedSum.Compress(0)
 		if err := readsummary.Set(ctx, batch, rec.GetRangeID(), ms, mergedSum); err != nil {
 			return result.Result{}, err
 		}

pkg/kv/kvserver/batcheval/cmd_lease.go

@@ -175,10 +175,13 @@ func evalNewLease(
 	// which case they could easily take a catastrophic turn) and the benefit is
 	// low.
 	if priorReadSum != nil {
-		if err := readsummary.Set(ctx, readWriter, rec.GetRangeID(), ms, priorReadSum); err != nil {
+		pd.Replicated.PriorReadSummary = priorReadSum
+		// Compress the persisted read summary, as it will likely never be needed.
+		compressedSum := priorReadSum.Clone()
+		compressedSum.Compress(0)
+		if err := readsummary.Set(ctx, readWriter, rec.GetRangeID(), ms, compressedSum); err != nil {
 			return newFailedLeaseTrigger(isTransfer), err
 		}
-		pd.Replicated.PriorReadSummary = priorReadSum
 	}
 
 	pd.Local.Metrics = new(result.Metrics)

pkg/kv/kvserver/batcheval/cmd_lease_test.go

@@ -15,6 +15,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/readsummary/rspb"
@@ -128,7 +129,15 @@ func TestLeaseTransferForwardsStartTime(t *testing.T) {
 			} else {
 				maxPriorReadTS = nextLease.Start.ToTimestamp().Add(-2*time.Second.Nanoseconds(), 0)
 			}
-			currentReadSummary := rspb.FromTimestamp(maxPriorReadTS)
+			currentReadSummary := rspb.FromTimestamp(maxPriorReadTS.Add(-100, 0))
+			currentReadSummary.Local.AddReadSpan(rspb.ReadSpan{
+				Key:       keys.MakeRangeKeyPrefix([]byte("a")),
+				Timestamp: maxPriorReadTS,
+			})
+			currentReadSummary.Global.AddReadSpan(rspb.ReadSpan{
+				Key:       roachpb.Key("a"),
+				Timestamp: maxPriorReadTS,
+			})
 
 			evalCtx := &MockEvalCtx{
 				ClusterSettings:    cluster.MakeTestingClusterSettings(),
@@ -165,11 +174,17 @@ func TestLeaseTransferForwardsStartTime(t *testing.T) {
 			// The prior read summary should reflect the maximum read times served
 			// under the current leaseholder, even if this time is below the proposed
 			// lease start time.
-			propReadSum, err := readsummary.Load(ctx, batch, desc.RangeID)
+			propReadSum := res.Replicated.PriorReadSummary
 			require.NoError(t, err)
-			require.NotNil(t, propReadSum, "should write prior read summary")
-			require.Equal(t, maxPriorReadTS, propReadSum.Local.LowWater)
-			require.Equal(t, maxPriorReadTS, propReadSum.Global.LowWater)
+			require.NotNil(t, propReadSum, "should include prior read summary")
+			require.Equal(t, currentReadSummary, *propReadSum)
+
+			// The prior read summary should also be persisted, but compressed.
+			persistReadSum, err := readsummary.Load(ctx, batch, desc.RangeID)
+			require.NoError(t, err)
+			require.NotNil(t, persistReadSum, "should persist prior read summary")
+			require.NotEqual(t, currentReadSummary, *persistReadSum)
+			require.Equal(t, rspb.FromTimestamp(maxPriorReadTS), *persistReadSum)
 		})
 	})
 }

pkg/kv/kvserver/client_merge_test.go

@@ -876,11 +876,7 @@ func mergeCheckingTimestampCaches(
 	ba.Add(hb)
 	var expReason kvpb.TransactionAbortedReason
 	if throughSnapshot {
-		// TODO(nvanbenschoten): this will result in the wrong reason until we
-		// start compressing the persisted read summary on lease transfers and
-		// range merges.
-		// expReason = kvpb.ABORT_REASON_TIMESTAMP_CACHE_REJECTED
-		expReason = kvpb.ABORT_REASON_ABORTED_RECORD_FOUND
+		expReason = kvpb.ABORT_REASON_TIMESTAMP_CACHE_REJECTED
 	} else {
 		expReason = kvpb.ABORT_REASON_ABORTED_RECORD_FOUND
 	}

pkg/kv/kvserver/readsummary/rspb/summary.go

@@ -47,6 +47,13 @@ func (c *ReadSummary) Merge(o ReadSummary) {
 	c.Global.Merge(o.Global)
 }
 
+// Compress compresses the read summary to fit within the provided size budget.
+// It splits the budget evenly between the local and global segments.
+func (c *ReadSummary) Compress(budget int64) {
+	c.Local.Compress(budget / 2)
+	c.Global.Compress(budget / 2)
+}
+
 // AddReadSpan adds a read span to the segment. The span must be sorted after
 // all existing spans in the segment and must not overlap with any existing
 // spans.