Merge pull request #4601 from nvanbenschoten/nvanbenschoten/mvccgc

storage/engine: Correctly handle inlined keys in MVCCGarbageCollect

Author: nvb

storage/engine/mvcc.go

@@ -1632,23 +1632,29 @@ func MVCCGarbageCollect(engine Engine, ms *MVCCStats, keys []roachpb.GCRequest_G
 			return err
 		}
 		if !ok {
-			return util.Errorf("could not seek to key %q", gcKey.Key)
+			continue
 		}
+		inlinedValue := meta.IsInline()
 		implicitMeta := iter.Key().IsValue()
 		// First, check whether all values of the key are being deleted.
 		if !gcKey.Timestamp.Less(meta.Timestamp) {
 			// For version keys, don't allow GC'ing the meta key if it's
 			// not marked deleted. However, for inline values we allow it;
 			// they are internal and GCing them directly saves the extra
 			// deletion step.
-			if !meta.Deleted && !gcKey.Timestamp.Equal(roachpb.ZeroTimestamp) {
+			if !meta.Deleted && !inlinedValue {
 				return util.Errorf("request to GC non-deleted, latest value of %q", gcKey.Key)
 			}
 			if meta.Txn != nil {
 				return util.Errorf("request to GC intent at %q", gcKey.Key)
 			}
 			if ms != nil {
-				ms.Add(updateStatsOnGC(gcKey.Key, metaKeySize, metaValSize, meta, meta.Timestamp.WallTime, timestamp.WallTime))
+				if inlinedValue {
+					updateStatsForInline(ms, gcKey.Key, metaKeySize, metaValSize, 0, 0)
+					ms.AgeTo(timestamp.WallTime)
+				} else {
+					ms.Add(updateStatsOnGC(gcKey.Key, metaKeySize, metaValSize, meta, meta.Timestamp.WallTime, timestamp.WallTime))
+				}
 			}
 			if !implicitMeta {
 				if err := engine.Clear(iter.Key()); err != nil {

storage/engine/mvcc_test.go

@@ -2645,6 +2645,7 @@ func TestMVCCGarbageCollect(t *testing.T) {
 	val1 := roachpb.MakeValueFromBytesAndTimestamp(bytes, ts1)
 	val2 := roachpb.MakeValueFromBytesAndTimestamp(bytes, ts2)
 	val3 := roachpb.MakeValueFromBytesAndTimestamp(bytes, ts3)
+	valInline := roachpb.MakeValueFromBytesAndTimestamp(bytes, roachpb.ZeroTimestamp)
 
 	testData := []struct {
 		key       roachpb.Key
@@ -2655,6 +2656,7 @@ func TestMVCCGarbageCollect(t *testing.T) {
 		{roachpb.Key("a-del"), []roachpb.Value{val1, val2}, true},
 		{roachpb.Key("b"), []roachpb.Value{val1, val2, val3}, false},
 		{roachpb.Key("b-del"), []roachpb.Value{val1, val2, val3}, true},
+		{roachpb.Key("inline"), []roachpb.Value{valInline}, false},
 	}
 
 	for i := 0; i < 3; i++ {
@@ -2677,12 +2679,12 @@ func TestMVCCGarbageCollect(t *testing.T) {
 			}
 		}
 	}
-	kvsn, err := Scan(engine, mvccKey(keyMin), mvccKey(keyMax), 0)
-	if err != nil {
-		t.Fatal(err)
-	}
-	for i, kv := range kvsn {
-		if log.V(1) {
+	if log.V(1) {
+		kvsn, err := Scan(engine, mvccKey(keyMin), mvccKey(keyMax), 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for i, kv := range kvsn {
 			log.Infof("%d: %s", i, kv.Key)
 		}
 	}
@@ -2692,6 +2694,10 @@ func TestMVCCGarbageCollect(t *testing.T) {
 		{Key: roachpb.Key("a-del"), Timestamp: ts2},
 		{Key: roachpb.Key("b"), Timestamp: ts1},
 		{Key: roachpb.Key("b-del"), Timestamp: ts2},
+		{Key: roachpb.Key("inline"), Timestamp: roachpb.ZeroTimestamp},
+		// Keys that don't exist, which should result in a no-op.
+		{Key: roachpb.Key("a-bad"), Timestamp: ts2},
+		{Key: roachpb.Key("inline-bad"), Timestamp: roachpb.ZeroTimestamp},
 	}
 	if err := MVCCGarbageCollect(engine, ms, keys, ts3); err != nil {
 		t.Fatal(err)
@@ -2761,21 +2767,40 @@ func TestMVCCGarbageCollectNonDeleted(t *testing.T) {
 	ts2 := makeTS(2E9, 0)
 	val1 := mkVal(s, ts1)
 	val2 := mkVal(s, ts2)
-	key := roachpb.Key("a")
-	vals := []roachpb.Value{val1, val2}
-	for _, val := range vals {
-		valCpy := *util.CloneProto(&val).(*roachpb.Value)
-		valCpy.Timestamp = roachpb.ZeroTimestamp
-		if err := MVCCPut(engine, nil, key, val.Timestamp, valCpy, nil); err != nil {
-			t.Fatal(err)
-		}
-	}
-	keys := []roachpb.GCRequest_GCKey{
-		{Key: roachpb.Key("a"), Timestamp: ts2},
+	valInline := mkVal(s, roachpb.ZeroTimestamp)
+
+	testData := []struct {
+		key      roachpb.Key
+		vals     []roachpb.Value
+		expError string
+	}{
+		{roachpb.Key("a"), []roachpb.Value{val1, val2}, `request to GC non-deleted, latest value ok "a"`},
+		{roachpb.Key("inline"), []roachpb.Value{valInline}, ""},
 	}
-	if err := MVCCGarbageCollect(engine, nil, keys, ts2); err == nil {
-		t.Fatal("expected error garbage collecting a non-deleted live value")
+
+	for _, test := range testData {
+		for _, val := range test.vals {
+			valCpy := *util.CloneProto(&val).(*roachpb.Value)
+			valCpy.Timestamp = roachpb.ZeroTimestamp
+			if err := MVCCPut(engine, nil, test.key, val.Timestamp, valCpy, nil); err != nil {
+				t.Fatal(err)
+			}
+		}
+		keys := []roachpb.GCRequest_GCKey{
+			{Key: test.key, Timestamp: ts2},
+		}
+		err := MVCCGarbageCollect(engine, nil, keys, ts2)
+		if test.expError == "" {
+			if err != nil {
+				t.Fatalf("expected no error garbage collecting a non-deleted inline live value, found %v", err)
+			}
+		} else {
+			if testutils.IsError(err, test.expError) {
+				t.Fatalf("expected error %q when garbage collecting a non-deleted live value, found %v", test.expError, err)
+			}
+		}
 	}
+
 }
 
 // TestMVCCGarbageCollectIntent verifies that an intent cannot be GC'd.
@@ -2800,7 +2825,7 @@ func TestMVCCGarbageCollectIntent(t *testing.T) {
 		t.Fatal(err)
 	}
 	keys := []roachpb.GCRequest_GCKey{
-		{Key: roachpb.Key("a"), Timestamp: ts2},
+		{Key: key, Timestamp: ts2},
 	}
 	if err := MVCCGarbageCollect(engine, nil, keys, ts2); err == nil {
 		t.Fatal("expected error garbage collecting an intent")