kv, roachpb: move seq num manipulation under Transaction

It belongs better there than as a standalone function, particularly
since we need to give special consideration to transaction fields (see
next commit).

Release note: None

pkg/kv/txn_coord_sender_savepoints.go

@@ -14,7 +14,6 @@ import (
 	"context"
 
 	"github.com/cockroachdb/cockroach/pkg/internal/client"
-	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
@@ -118,8 +117,7 @@ func (tc *TxnCoordSender) RollbackToSavepoint(ctx context.Context, s client.Save
 		return nil
 	}
 
-	tc.mu.txn.IgnoredSeqNums = roachpb.AddIgnoredSeqNumRange(
-		tc.mu.txn.IgnoredSeqNums,
+	tc.mu.txn.AddIgnoredSeqNumRange(
 		enginepb.IgnoredSeqNumRange{
 			Start: st.seqNum + 1, End: tc.interceptorAlloc.txnSeqNumAllocator.writeSeq,
 		})

pkg/roachpb/data.go

@@ -1162,6 +1162,52 @@ func (t *Transaction) GetObservedTimestamp(nodeID NodeID) (hlc.Timestamp, bool)
 	return s.get(nodeID)
 }
 
+// AddIgnoredSeqNumRange adds the given range to the transaction's list of
+// ignored seqnum ranges.
+//
+// The following invariants are assumed to hold and are preserved:
+// - the list contains no overlapping ranges
+// - the list contains no contiguous ranges
+// - the list is sorted, with larger seqnums at the end
+//
+// Additionally, the caller must ensure:
+//
+// 1) if the new range overlaps with some range in the list, then it
+//    also overlaps with every subsequent range in the list.
+//
+// 2) the new range's "end" seqnum is larger or equal to the "end"
+//    seqnum of the last element in the list.
+//
+// For example:
+//     current list [3 5] [10 20] [22 24]
+//     new item:    [8 26]
+//     final list:  [3 5] [8 26]
+//
+//     current list [3 5] [10 20] [22 24]
+//     new item:    [28 32]
+//     final list:  [3 5] [10 20] [22 24] [28 32]
+//
+// This corresponds to savepoints semantics:
+//
+// - Property 1 says that a rollback to an earlier savepoint
+//   rolls back over all writes following that savepoint.
+// - Property 2 comes from that the new range's 'end' seqnum is the
+//   current write seqnum and thus larger than or equal to every
+//   previously seen value.
+func (t *Transaction) AddIgnoredSeqNumRange(newRange enginepb.IgnoredSeqNumRange) {
+	// Truncate the list at the last element not included in the new range.
+	list := t.IgnoredSeqNums
+	i := 0
+	for ; i < len(list); i++ {
+		if list[i].End < newRange.Start {
+			continue
+		}
+		break
+	}
+	list = list[:i]
+	t.IgnoredSeqNums = append(list, newRange)
+}
+
 // AsRecord returns a TransactionRecord object containing only the subset of
 // fields from the receiver that must be persisted in the transaction record.
 func (t *Transaction) AsRecord() TransactionRecord {

pkg/roachpb/data_test.go

@@ -1935,3 +1935,41 @@ func TestAsLockUpdates(t *testing.T) {
 		require.Equal(t, dur, intent.Durability)
 	}
 }
+
+func TestAddIgnoredSeqNumRange(t *testing.T) {
+	type r = enginepb.IgnoredSeqNumRange
+
+	mr := func(a, b enginepb.TxnSeq) r {
+		return r{Start: a, End: b}
+	}
+
+	testData := []struct {
+		list     []r
+		newRange r
+		exp      []r
+	}{
+		{[]r{},
+			mr(1, 2),
+			[]r{mr(1, 2)}},
+		{[]r{mr(1, 2)},
+			mr(1, 4),
+			[]r{mr(1, 4)}},
+		{[]r{mr(1, 2), mr(3, 6)},
+			mr(8, 10),
+			[]r{mr(1, 2), mr(3, 6), mr(8, 10)}},
+		{[]r{mr(1, 2), mr(5, 6)},
+			mr(3, 8),
+			[]r{mr(1, 2), mr(3, 8)}},
+		{[]r{mr(1, 2), mr(5, 6)},
+			mr(1, 8),
+			[]r{mr(1, 8)}},
+	}
+
+	for _, tc := range testData {
+		txn := Transaction{
+			IgnoredSeqNums: tc.list,
+		}
+		txn.AddIgnoredSeqNumRange(tc.newRange)
+		require.Equal(t, tc.exp, txn.IgnoredSeqNums)
+	}
+}