1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
//
// Test to verify that updates that would change the resharding key value are replicated as an
// insert, delete pair.
// @tags: [
// uses_atclustertime,
// ]
//
(function() {
'use strict';
load('jstests/libs/discover_topology.js');
load('jstests/sharding/libs/resharding_test_fixture.js');
load('jstests/sharding/libs/sharded_transactions_helpers.js');
const reshardingTest = new ReshardingTest({numDonors: 2, numRecipients: 2, reshardInPlace: true});
reshardingTest.setup();
const donorShardNames = reshardingTest.donorShardNames;
const testColl = reshardingTest.createShardedCollection({
ns: 'test.foo',
shardKeyPattern: {x: 1, s: 1},
chunks: [
{min: {x: MinKey, s: MinKey}, max: {x: 5, s: 5}, shard: donorShardNames[0]},
{min: {x: 5, s: 5}, max: {x: MaxKey, s: MaxKey}, shard: donorShardNames[1]},
],
});
const docToUpdate = ({_id: 0, x: 2, s: 2, y: 2});
assert.commandWorked(testColl.insert(docToUpdate));
let retryableWriteTs;
let txnWriteTs;
const mongos = testColl.getMongo();
const updateDocumentShardKeyUsingTransactionApiEnabled =
isUpdateDocumentShardKeyUsingTransactionApiEnabled(mongos);
const recipientShardNames = reshardingTest.recipientShardNames;
reshardingTest.withReshardingInBackground( //
{
newShardKeyPattern: {y: 1, s: 1},
newChunks: [
{min: {y: MinKey, s: MinKey}, max: {y: 5, s: 5}, shard: recipientShardNames[0]},
{min: {y: 5, s: 5}, max: {y: MaxKey, s: MaxKey}, shard: recipientShardNames[1]},
],
},
(tempNs) => {
// Wait for cloning to have at least started on the recipient shards to know that the donor
// shards have begun including the "destinedRecipient" field in their oplog entries.
const tempColl = mongos.getCollection(tempNs);
assert.soon(() => tempColl.findOne(docToUpdate) !== null);
// When the updateDocumentShardKeyUsingTransactionApi feature flag is enabled, ordinary
// updates that modify a document's shard key will complete.
assert.commandWorked(testColl.insert({_id: 1, x: 2, s: 2, y: 2}));
const updateRes = testColl.update({_id: 1, x: 2, s: 2}, {$set: {y: 10}});
if (updateDocumentShardKeyUsingTransactionApiEnabled) {
assert.commandWorked(updateRes);
} else {
assert.commandFailedWithCode(
updateRes,
ErrorCodes.IllegalOperation,
'was able to update value under new shard key as ordinary write');
}
const session = testColl.getMongo().startSession({retryWrites: true});
const sessionColl =
session.getDatabase(testColl.getDB().getName()).getCollection(testColl.getName());
assert.commandFailedWithCode(
sessionColl.update({_id: 0, x: 2, s: 2}, {$set: {y: 10}}, {multi: true}),
ErrorCodes.InvalidOptions,
'was able to update value under new shard key when {multi: true} specified');
assert.commandFailedWithCode(
sessionColl.update({_id: 0}, {$set: {y: 10}}),
31025,
'was able to update value under new shard key without specifying the full shard ' +
'key in the query');
let res;
assert.soon(
() => {
res = sessionColl.update({_id: 0, x: 2, s: 2}, {$set: {y: 20, s: 3}});
if (res.nModified === 1) {
assert.commandWorked(res);
retryableWriteTs = session.getOperationTime();
return true;
}
assert.commandFailedWithCode(res, [
ErrorCodes.StaleConfig,
ErrorCodes.NoSuchTransaction,
ErrorCodes.ShardCannotRefreshDueToLocksHeld
]);
return false;
},
() => `was unable to update value under new shard key as retryable write: ${
tojson(res)}`);
assert.soon(() => {
session.startTransaction();
res = sessionColl.update({_id: 0, x: 2, s: 3}, {$set: {y: -30, s: 1}});
if (res.nModified === 1) {
session.commitTransaction();
txnWriteTs = session.getOperationTime();
return true;
}
// mongos will automatically retry the update as a pair of delete and insert commands in
// a multi-document transaction. We permit NoSuchTransaction errors because it is
// possible for the resharding operation running in the background to cause the shard
// version to be bumped. The StaleConfig error won't be automatically retried by mongos
// for the second statement in the transaction (the insert) and would lead to a
// NoSuchTransaction error.
if (updateDocumentShardKeyUsingTransactionApiEnabled) {
// The handling of WCOS errors with internal transactions advances the router's
// notion of the transaction "statement" number between the initial update, the
// delete, and the insert, so if the shard version changes and is detected by the
// delete or insert, the router will refuse to retry and return the stale version
// error instead.
assert.commandFailedWithCode(res, [
ErrorCodes.NoSuchTransaction,
ErrorCodes.ShardCannotRefreshDueToLocksHeld,
ErrorCodes.NoSuchTransaction
]);
} else {
assert.commandFailedWithCode(res, ErrorCodes.NoSuchTransaction);
}
session.abortTransaction();
return false;
}, () => `was unable to update value under new shard key in transaction: ${tojson(res)}`);
});
const topology = DiscoverTopology.findConnectedNodes(mongos);
const donor0 = new Mongo(topology.shards[donorShardNames[0]].primary);
const donorOplogColl0 = donor0.getCollection('local.oplog.rs');
function assertOplogEntryIsDeleteInsertApplyOps(entry, isRetryableWrite) {
assert(entry.o.hasOwnProperty('applyOps'), entry);
if (updateDocumentShardKeyUsingTransactionApiEnabled && isRetryableWrite) {
// With internal transactions the applyOps array for a retryable write update will have a
// noop entry at the front.
assert.eq(entry.o.applyOps.length, 3, entry);
assert.eq(entry.o.applyOps[0].op, 'n', entry);
assert.eq(entry.o.applyOps[1].op, 'd', entry);
assert.eq(entry.o.applyOps[1].ns, testColl.getFullName(), entry);
assert.eq(entry.o.applyOps[2].op, 'i', entry);
assert.eq(entry.o.applyOps[2].ns, testColl.getFullName(), entry);
} else {
assert.eq(entry.o.applyOps.length, 2, entry);
assert.eq(entry.o.applyOps[0].op, 'd', entry);
assert.eq(entry.o.applyOps[0].ns, testColl.getFullName(), entry);
assert.eq(entry.o.applyOps[1].op, 'i', entry);
assert.eq(entry.o.applyOps[1].ns, testColl.getFullName(), entry);
}
}
const retryableWriteEntry = donorOplogColl0.findOne({ts: retryableWriteTs});
assert.neq(null, retryableWriteEntry, 'failed to find oplog entry for retryable write');
assertOplogEntryIsDeleteInsertApplyOps(retryableWriteEntry, true /* isRetryableWrite */);
const txnWriteEntry = donorOplogColl0.findOne({ts: txnWriteTs});
assert.neq(null, txnWriteEntry, 'failed to find oplog entry for transaction');
assertOplogEntryIsDeleteInsertApplyOps(txnWriteEntry);
reshardingTest.teardown();
})();
|
