From c203c826182546a861d1b1cb2d4694f0dedcb38b Mon Sep 17 00:00:00 2001
From: Bilal Akhtar <bilal@cockroachlabs.com>
Date: Fri, 31 Mar 2023 19:42:47 +0000
Subject: [PATCH] storage: CheckSSTConflicts stats fixes

Fixes some additional cases of stats divergence
in CheckSSTConflicts' handling of inbound sst range key
fragments that shadow points in engine and fragment
existing engine range keys.

Found by randomized test in #98408.

Epic: none

Release note: None
---
 .../batcheval/cmd_add_sstable_test.go         | 24 +++++++++++++++++++
 pkg/storage/sst.go                            |  9 ++++---
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go b/pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go
index 813305964774..f5682df778ee 100644
--- a/pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go
+++ b/pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go
@@ -1089,6 +1089,30 @@ func TestEvalAddSSTable(t *testing.T) {
 			sst:          kvs{rangeKV("a", "b", 10, ""), pointKV("d", 9, "foo")},
 			ignoreExpect: true,
 		},
+		"DisallowConflict does not double count deleted ext key": {
+			noConflict:   true,
+			data:         kvs{pointKV("e", 6, "foo"), rangeKV("e", "g", 5, "")},
+			sst:          kvs{rangeKV("a", "j", 10, ""), pointKV("b", 11, "bar"), pointKV("c", 11, "foo"), pointKV("d", 11, "foo"), pointKV("f", 11, "foo")},
+			ignoreExpect: true,
+		},
+		"DisallowConflict does not double count deleted ext key 2": {
+			noConflict:   true,
+			data:         kvs{pointKV("b", 7, "foo"), pointKV("d", 6, "foo"), rangeKV("d", "e", 5, "")},
+			sst:          kvs{rangeKV("a", "j", 10, ""), pointKV("c", 11, "foo"), pointKV("d", 12, "bar")},
+			ignoreExpect: true,
+		},
+		"DisallowConflict handles complex range key cases": {
+			noConflict:   true,
+			data:         kvs{pointKV("cb", 6, "foo"), pointKV("cm", 6, "foo"), pointKV("cn", 6, ""), pointKV("co", 6, "foo"), rangeKV("c", "co", 5, "")},
+			sst:          kvs{rangeKV("cd", "cnr", 9, ""), pointKV("cn", 8, "bar"), rangeKV("cnr", "d", 10, ""), pointKV("co", 8, "bar")},
+			ignoreExpect: true,
+		},
+		"DisallowConflict handles complex range key cases 2": {
+			noConflict:   true,
+			data:         kvs{pointKV("cb", 6, "foo"), pointKV("cm", 6, "foo"), pointKV("cn", 6, ""), pointKV("cx", 6, "foo"), rangeKV("c", "co", 5, "")},
+			sst:          kvs{rangeKV("cd", "cnr", 9, ""), pointKV("cn", 8, "bar"), rangeKV("cnr", "d", 10, ""), pointKV("co", 8, "bar")},
+			ignoreExpect: true,
+		},
 	}
 	testutils.RunTrueAndFalse(t, "IngestAsWrites", func(t *testing.T, ingestAsWrites bool) {
 		testutils.RunValues(t, "RewriteConcurrency", []interface{}{0, 8}, func(t *testing.T, c interface{}) {
diff --git a/pkg/storage/sst.go b/pkg/storage/sst.go
index 9089eed401a4..2834adc76d88 100644
--- a/pkg/storage/sst.go
+++ b/pkg/storage/sst.go
@@ -375,7 +375,7 @@ func CheckSSTConflicts(
 	var extErr error
 	var sstPrevRangeKeys, extPrevRangeKeys MVCCRangeKeyStack
 	var sstFirstRangeKey MVCCRangeKeyStack
-	var extPrevKey MVCCKey
+	var extPrevKey, extPrevDeletedKey MVCCKey
 
 	if usePrefixSeek {
 		// In the case of prefix seeks, do not look at engine iter exhaustion. This
@@ -551,10 +551,11 @@ func CheckSSTConflicts(
 						return enginepb.MVCCStats{}, errors.AssertionFailedf("expected range tombstone above timestamp %v", extKey.Timestamp)
 					}
 					sstPointShadowsExtPoint := sstHasPoint && sstIter.UnsafeKey().Key.Equal(extKey.Key)
-					if (extKeyChanged || sstRangeKeysChanged) && !sstPointShadowsExtPoint {
+					if (extKeyChanged || sstRangeKeysChanged) && !sstPointShadowsExtPoint && !extKey.Equal(extPrevDeletedKey) {
+						extKey.CloneInto(&extPrevDeletedKey)
 						statsDiff.Add(updateStatsOnRangeKeyCover(
 							sstRangeKeyVersion.Timestamp, extKey, extValueLen, extValueIsTombstone))
-					} else if !extKeyChanged && sstPointShadowsExtPoint {
+					} else if extKey.Equal(extPrevDeletedKey) && sstPointShadowsExtPoint {
 						// This is either a conflict, shadow, or idempotent operation.
 						// Subtract the RangeKeyCover stats diff from the last iteration, as
 						// compareForCollision will account for the shadow.
@@ -993,6 +994,8 @@ func CheckSSTConflicts(
 					sstOK, sstErr = sstIter.Valid()
 					extIter.SeekGE(extPrevKey)
 					extOK, extErr = extIter.Valid()
+					// We could have reset extHasRange above, so set it back.
+					_, extHasRange = extIter.HasPointAndRange()
 				}
 			}
 		}