storage: error MVCCExportToSST when a single KV is larger than the ma…

…x size

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 detection 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

pkg/storage/mvcc.go

@@ -4869,9 +4869,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)
+}