Merge #83102

83102: storage: return error from MVCCExportToSST when a KV is larger than the max size r=stevendanna a=stevendanna

Previously, if a single KV was larger than the max size,
MVCCExportToSST would not add the KV to the SST but also return a nil
error. This was because the current code assumed that we
might be able to resume. So rather than failing, it set the resume
key. Then, before exiting, if the SST is empty, we early return an
empty resume key and a nil error.

Thus, anytime we hit a key larger than the max as the first key being
added to the SST, we would silently ignore it.

Now, we check for this condition early and return an error.

Fixes #83100

Release note (bug fix): Fix bug where BACKUP may be missing data when
the cluster was configured with a very low values for
kv.bulk_sst.max_allowed_overage and kv.bulk_sst.target_size

Co-authored-by: Steven Danna <[email protected]>

pkg/storage/mvcc.go

@@ -4921,9 +4921,14 @@ func MVCCExportToSST(
 			reachedTargetSize := curSizeWithRangeKeys > 0 &&
 				uint64(curSizeWithRangeKeys) >= opts.TargetSize
 			kvSize := int64(len(unsafeKey.Key) + len(unsafeValue))
+			if curSize == 0 && opts.MaxSize > 0 && kvSize > int64(opts.MaxSize) {
+				// This single key exceeds the MaxSize. Even if we paginate below, this will still fail.
+				return roachpb.BulkOpSummary{}, MVCCKey{}, &ExceedMaxSizeError{reached: kvSize, maxSize: opts.MaxSize}
+			}
 			newSize := curSize + kvSize
 			newSizeWithRangeKeys := curSizeWithRangeKeys + kvSize
 			reachedMaxSize := opts.MaxSize > 0 && newSizeWithRangeKeys > int64(opts.MaxSize)
+
 			// When paginating we stop writing in two cases:
 			// - target size is reached and we wrote all versions of a key
 			// - maximum size reached and we are allowed to stop mid key

pkg/storage/mvcc_test.go

@@ -5684,3 +5684,31 @@ func TestMVCCExportToSSTSplitMidKey(t *testing.T) {
 			})
 	}
 }
+
+// TestMVCCExportToSSTSErrorsOnLargeKV verifies that MVCCExportToSST errors on a
+// single kv that is larger than max size.
+func TestMVCCExportToSSTSErrorsOnLargeKV(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	st := cluster.MakeTestingClusterSettings()
+
+	engine := createTestPebbleEngine()
+	defer engine.Close()
+	var testData = []testValue{value(key(1), "value1", ts(1000))}
+	require.NoError(t, fillInData(ctx, engine, testData))
+	summary, _, err := MVCCExportToSST(
+		ctx, st, engine, MVCCExportOptions{
+			StartKey:           MVCCKey{Key: key(1)},
+			EndKey:             key(3).Next(),
+			StartTS:            hlc.Timestamp{},
+			EndTS:              hlc.Timestamp{WallTime: 9999},
+			ExportAllRevisions: false,
+			TargetSize:         1,
+			MaxSize:            1,
+			StopMidKey:         true,
+		}, &MemFile{})
+	require.Equal(t, int64(0), summary.DataSize)
+	expectedErr := &ExceedMaxSizeError{}
+	require.ErrorAs(t, err, &expectedErr)
+}