Skip to content

Commit a8ead97

Browse files
stevendannastevendanna
committed
kvcoord: make write buffer options metamorphic
This makes the following options metamorphic: kv.transaction.write_buffering.transformations.scans.enabled kv.transaction.write_buffering.max_buffer_size Epic: none Release note: None
1 parent de90bde commit a8ead97

File tree

2 files changed

+46
-41
lines changed

2 files changed

+46
-41
lines changed

‎pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,15 +48,20 @@ var bufferedWritesScanTransformEnabled = settings.RegisterBoolSetting(
4848
settings.ApplicationLevel,
4949
"kv.transaction.write_buffering.transformations.scans.enabled",
5050
"if enabled, locking scans and reverse scans with replicated durability are transformed to unreplicated durability",
51-
true,
51+
metamorphic.ConstantWithTestBool("kv.transaction.write_buffering.transformations.scans.enabled", true /* defaultValue */),
5252
)
5353

54+
const defaultBufferSize = 1 << 22 // 4MB
5455
var bufferedWritesMaxBufferSize = settings.RegisterByteSizeSetting(
5556
settings.ApplicationLevel,
5657
"kv.transaction.write_buffering.max_buffer_size",
5758
"if non-zero, defines that maximum size of the "+
5859
"buffer that will be used to buffer transactional writes per-transaction",
59-
1<<22, // 4MB
60+
int64(metamorphic.ConstantWithTestRange("kv.transaction.write_buffering.max_buffer_size",
61+
defaultBufferSize, // default
62+
1, // min
63+
defaultBufferSize, // max
64+
)),
6065
settings.NonNegativeInt,
6166
settings.WithPublic,
6267
)

‎pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go‎

Lines changed: 39 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,12 @@ import (
2727
)
2828

2929
func makeMockTxnWriteBuffer(
30-
st *cluster.Settings, optionalMetrics ...TxnMetrics,
31-
) (txnWriteBuffer, *mockLockedSender) {
30+
ctx context.Context, optionalMetrics ...TxnMetrics,
31+
) (txnWriteBuffer, *mockLockedSender, *cluster.Settings) {
32+
st := cluster.MakeClusterSettings()
33+
bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, true)
34+
bufferedWritesMaxBufferSize.Override(ctx, &st.SV, defaultBufferSize)
35+
3236
var metrics TxnMetrics
3337
if len(optionalMetrics) > 0 {
3438
metrics = optionalMetrics[0]
@@ -41,7 +45,7 @@ func makeMockTxnWriteBuffer(
4145
wrapped: mockSender,
4246
txnMetrics: &metrics,
4347
st: st,
44-
}, mockSender
48+
}, mockSender, st
4549
}
4650

4751
func getArgs(key roachpb.Key) *kvpb.GetRequest {
@@ -99,7 +103,7 @@ func TestTxnWriteBufferBuffersBlindWrites(t *testing.T) {
99103
defer leaktest.AfterTest(t)()
100104
defer log.Scope(t).Close(t)
101105
ctx := context.Background()
102-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
106+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
103107

104108
txn := makeTxnProto()
105109
txn.Sequence = 1
@@ -172,7 +176,7 @@ func TestTxnWriteBufferWritesToSameKey(t *testing.T) {
172176
defer leaktest.AfterTest(t)()
173177
defer log.Scope(t).Close(t)
174178
ctx := context.Background()
175-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
179+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
176180

177181
txn := makeTxnProto()
178182
txn.Sequence = 1
@@ -270,7 +274,7 @@ func TestTxnWriteBufferBlindWritesIncludingOtherRequests(t *testing.T) {
270274
defer leaktest.AfterTest(t)()
271275
defer log.Scope(t).Close(t)
272276
ctx := context.Background()
273-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
277+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
274278

275279
txn := makeTxnProto()
276280
txn.Sequence = 1
@@ -365,7 +369,7 @@ func TestTxnWriteBufferCorrectlyAdjustsFlushErrors(t *testing.T) {
365369
} {
366370
t.Run(fmt.Sprintf("errIdx=%d", errIdx), func(t *testing.T) {
367371
ctx := context.Background()
368-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
372+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
369373

370374
txn := makeTxnProto()
371375
txn.Sequence = 1
@@ -446,7 +450,7 @@ func TestTxnWriteBufferCorrectlyAdjustsErrorsAfterBuffering(t *testing.T) {
446450
} {
447451
t.Run(fmt.Sprintf("errIdx=%d", errIdx), func(t *testing.T) {
448452
ctx := context.Background()
449-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
453+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
450454

451455
txn := makeTxnProto()
452456
txn.Sequence = 1
@@ -536,7 +540,7 @@ func TestTxnWriteBufferServesPointReadsLocally(t *testing.T) {
536540
defer leaktest.AfterTest(t)()
537541
defer log.Scope(t).Close(t)
538542
ctx := context.Background()
539-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
543+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
540544

541545
putAtSeq := func(key roachpb.Key, val string, seq enginepb.TxnSeq) {
542546
txn := makeTxnProto()
@@ -730,7 +734,7 @@ func TestTxnWriteBufferServesPointReadsAfterScan(t *testing.T) {
730734
defer leaktest.AfterTest(t)()
731735
defer log.Scope(t).Close(t)
732736
ctx := context.Background()
733-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
737+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
734738

735739
txn := makeTxnProto()
736740
txn.Sequence = 10
@@ -790,7 +794,7 @@ func TestTxnWriteBufferServesOverlappingReadsCorrectly(t *testing.T) {
790794
defer leaktest.AfterTest(t)()
791795
defer log.Scope(t).Close(t)
792796
ctx := context.Background()
793-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
797+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
794798

795799
putAtSeq := func(key roachpb.Key, val string, seq enginepb.TxnSeq) {
796800
txn := makeTxnProto()
@@ -1015,7 +1019,7 @@ func TestTxnWriteBufferLockingGetRequests(t *testing.T) {
10151019
defer leaktest.AfterTest(t)()
10161020
defer log.Scope(t).Close(t)
10171021
ctx := context.Background()
1018-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1022+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
10191023

10201024
txn := makeTxnProto()
10211025
txn.Sequence = 10
@@ -1123,7 +1127,7 @@ func TestTxnWriteBufferDecomposesConditionalPuts(t *testing.T) {
11231127

11241128
testutils.RunTrueAndFalse(t, "condEvalSuccessful", func(t *testing.T, condEvalSuccessful bool) {
11251129
ctx := context.Background()
1126-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1130+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
11271131
twb.testingOverrideCPutEvalFn = func(expBytes []byte, actVal *roachpb.Value, actValPresent bool, allowNoExisting bool) *kvpb.ConditionFailedError {
11281132
if condEvalSuccessful {
11291133
return nil
@@ -1203,7 +1207,7 @@ func TestTxnWriteBufferDecomposesConditionalPutsExpectingNoRow(t *testing.T) {
12031207
defer leaktest.AfterTest(t)()
12041208
defer log.Scope(t).Close(t)
12051209
ctx := context.Background()
1206-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1210+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
12071211

12081212
twb.testingOverrideCPutEvalFn = func(expBytes []byte, actVal *roachpb.Value, actValPresent bool, allowNoExisting bool) *kvpb.ConditionFailedError {
12091213
return nil
@@ -1265,7 +1269,7 @@ func TestTxnWriteBufferRespectsMustAcquireExclusiveLock(t *testing.T) {
12651269
defer leaktest.AfterTest(t)()
12661270
defer log.Scope(t).Close(t)
12671271
ctx := context.Background()
1268-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1272+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
12691273

12701274
txn := makeTxnProto()
12711275
txn.Sequence = 10
@@ -1338,7 +1342,7 @@ func TestTxnWriteBufferMustSortBatchesBySequenceNumber(t *testing.T) {
13381342
defer leaktest.AfterTest(t)()
13391343
defer log.Scope(t).Close(t)
13401344
ctx := context.Background()
1341-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1345+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
13421346

13431347
txn := makeTxnProto()
13441348
txn.Sequence = 10
@@ -1394,10 +1398,9 @@ func TestTxnWriteBufferMustSortBatchesBySequenceNumber(t *testing.T) {
13941398
func TestTxnWriteBufferEstimateSize(t *testing.T) {
13951399
defer leaktest.AfterTest(t)()
13961400
defer log.Scope(t).Close(t)
1397-
twb, _ := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1401+
ctx := context.Background()
13981402

1399-
st := cluster.MakeTestingClusterSettings()
1400-
twb.st = st
1403+
twb, _, _ := makeMockTxnWriteBuffer(ctx)
14011404

14021405
txn := makeTxnProto()
14031406
txn.Sequence = 10
@@ -1497,8 +1500,7 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
14971500
defer leaktest.AfterTest(t)()
14981501
defer log.Scope(t).Close(t)
14991502
ctx := context.Background()
1500-
st := cluster.MakeTestingClusterSettings()
1501-
twb, mockSender := makeMockTxnWriteBuffer(st)
1503+
twb, mockSender, st := makeMockTxnWriteBuffer(ctx)
15021504

15031505
txn := makeTxnProto()
15041506
txn.Sequence = 10
@@ -1615,7 +1617,6 @@ func TestTxnWriteBufferLimitsSizeOfScans(t *testing.T) {
16151617
defer log.Scope(t).Close(t)
16161618

16171619
ctx := context.Background()
1618-
st := cluster.MakeTestingClusterSettings()
16191620

16201621
keyA, keyB := roachpb.Key("a"), roachpb.Key("b")
16211622

@@ -1657,10 +1658,11 @@ func TestTxnWriteBufferLimitsSizeOfScans(t *testing.T) {
16571658
}
16581659
name := fmt.Sprintf("%s/%s", req, tc.name)
16591660
t.Run(name, func(t *testing.T) {
1660-
twb, mockSender := makeMockTxnWriteBuffer(st)
1661+
twb, mockSender, st := makeMockTxnWriteBuffer(ctx)
16611662
txn := makeTxnProto()
16621663
txn.Sequence = 10
16631664

1665+
bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, true)
16641666
bufferedWritesMaxBufferSize.Override(ctx, &st.SV, tc.bufferSize)
16651667

16661668
ba := &kvpb.BatchRequest{Header: kvpb.Header{Txn: &txn}}
@@ -1695,7 +1697,7 @@ func TestTxnWriteBufferLimitsSizeOfScans(t *testing.T) {
16951697
}
16961698
}
16971699
t.Run("no mutation when an unsupported request is in batch", func(t *testing.T) {
1698-
twb, mockSender := makeMockTxnWriteBuffer(st)
1700+
twb, mockSender, st := makeMockTxnWriteBuffer(ctx)
16991701
txn := makeTxnProto()
17001702
txn.Sequence = 10
17011703

@@ -1799,7 +1801,7 @@ func TestTxnWriteBufferFlushesIfBatchRequiresFlushing(t *testing.T) {
17991801

18001802
for _, tc := range testCases {
18011803
t.Run(tc.name, func(t *testing.T) {
1802-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1804+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
18031805

18041806
txn := makeTxnProto()
18051807
txn.Sequence = 10
@@ -1889,7 +1891,7 @@ func TestTxnWriteBufferRollbackToSavepoint(t *testing.T) {
18891891
defer leaktest.AfterTest(t)()
18901892
defer log.Scope(t).Close(t)
18911893
ctx := context.Background()
1892-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
1894+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
18931895

18941896
txn := makeTxnProto()
18951897
txn.Sequence = 10
@@ -1997,7 +1999,7 @@ func TestRollbackNeverHeldLock(t *testing.T) {
19971999
defer leaktest.AfterTest(t)()
19982000
defer log.Scope(t).Close(t)
19992001
ctx := context.Background()
2000-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2002+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
20012003

20022004
txn := makeTxnProto()
20032005
txn.Sequence = 10
@@ -2037,7 +2039,7 @@ func TestTxnWriteBufferFlushesAfterDisabling(t *testing.T) {
20372039
defer leaktest.AfterTest(t)()
20382040
defer log.Scope(t).Close(t)
20392041
ctx := context.Background()
2040-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2042+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
20412043

20422044
txn := makeTxnProto()
20432045
txn.Sequence = 1
@@ -2145,7 +2147,7 @@ func TestTxnWriteBufferClearsBufferOnEpochBump(t *testing.T) {
21452147
defer leaktest.AfterTest(t)()
21462148
defer log.Scope(t).Close(t)
21472149
ctx := context.Background()
2148-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2150+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
21492151

21502152
txn := makeTxnProto()
21512153
txn.Sequence = 1
@@ -2179,7 +2181,7 @@ func TestTxnWriteBufferBatchRequestValidation(t *testing.T) {
21792181
defer leaktest.AfterTest(t)()
21802182
defer log.Scope(t).Close(t)
21812183
ctx := context.Background()
2182-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2184+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
21832185

21842186
type testCase struct {
21852187
name string
@@ -2457,8 +2459,7 @@ func TestTxnWriteBufferHasBufferedAllPrecedingWrites(t *testing.T) {
24572459
} {
24582460
t.Run(tc.name, func(t *testing.T) {
24592461
ctx := context.Background()
2460-
st := cluster.MakeTestingClusterSettings()
2461-
twb, mockSender := makeMockTxnWriteBuffer(st)
2462+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
24622463

24632464
if tc.setup != nil {
24642465
tc.setup(&twb)
@@ -2506,7 +2507,6 @@ func TestTxnWriteBufferHasBufferedAllPrecedingWrites(t *testing.T) {
25062507
func BenchmarkTxnWriteBuffer(b *testing.B) {
25072508
defer leaktest.AfterTest(b)()
25082509
ctx := context.Background()
2509-
ct := cluster.MakeClusterSettings()
25102510
metrics := MakeTxnMetrics(time.Hour)
25112511

25122512
// Map from kvSize to a slice of keys where the i-th element corresponds to
@@ -2546,7 +2546,7 @@ func BenchmarkTxnWriteBuffer(b *testing.B) {
25462546
}
25472547
}
25482548
makeBuffer := func(kvSize int, txn *roachpb.Transaction, numWrites int) txnWriteBuffer {
2549-
twb, mockSender := makeMockTxnWriteBuffer(ct, metrics)
2549+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx, metrics)
25502550
sendFunc := func(ba *kvpb.BatchRequest) (*kvpb.BatchResponse, *kvpb.Error) {
25512551
br := ba.CreateReply()
25522552
br.Txn = ba.Txn
@@ -2700,7 +2700,7 @@ func TestTxnWriteBufferChecksForExclusionLoss(t *testing.T) {
27002700
defer log.Scope(t).Close(t)
27012701

27022702
ctx := context.Background()
2703-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2703+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
27042704

27052705
txn := makeTxnProto()
27062706
txn.Sequence = 10
@@ -2804,7 +2804,7 @@ func TestTxnWriteBufferCorrectlyRollsbackExclusionTimestamp(t *testing.T) {
28042804
defer log.Scope(t).Close(t)
28052805

28062806
ctx := context.Background()
2807-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
2807+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
28082808

28092809
txn := makeTxnProto()
28102810
txn.Sequence = 10
@@ -3242,7 +3242,7 @@ func TestTxnWriteBufferElidesUnnecessaryLockingRequests(t *testing.T) {
32423242
skip.WithIssue(t, 142977, "%s requires a value but %s does not buffer its response", firstReq.name, secondReq.name)
32433243
}
32443244

3245-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
3245+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
32463246
txn := makeTxnProto()
32473247
txn.Sequence = 10
32483248
// Send first request and run firstRequest validation
@@ -3490,7 +3490,7 @@ func TestTxnWriteBufferLockingReadsTransformations(t *testing.T) {
34903490

34913491
for _, tc := range testCases {
34923492
t.Run(tc.name, func(t *testing.T) {
3493-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
3493+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
34943494
txn := makeTxnProto()
34953495
txn.Sequence = 10
34963496

@@ -3712,7 +3712,7 @@ func TestTxnWriteBufferLockingGetFlushing(t *testing.T) {
37123712
for _, tc := range testCases {
37133713
name := strings.Join(tc.ops, "_")
37143714
t.Run(name, func(t *testing.T) {
3715-
twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
3715+
twb, mockSender, _ := makeMockTxnWriteBuffer(ctx)
37163716
txn := makeTxnProto()
37173717
txn.Sequence = 10
37183718
mockSender.MockSend(func(ba *kvpb.BatchRequest) (*kvpb.BatchResponse, *kvpb.Error) {

0 commit comments

Comments
 (0)