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
|
/**
* Test that optimized collection scan on the oplog collection can be applied and executed
* successfully.
* @tags: [
* requires_replication,
* ]
*/
(function() {
"use strict";
const rst = new ReplSetTest({nodes: 2});
rst.startSet();
rst.initiate();
const db = rst.getPrimary().getDB("oplog_scan_optimizations");
const localDb = rst.getPrimary().getDB("local");
const collName = "oplog_scan_optimizations";
const oplogCollName = "oplog.rs";
const coll = db[collName];
const oplogColl = localDb[oplogCollName];
coll.drop();
// Insert several document so that we can use a cursor to fetch them in multiple batches.
const testData = [{_id: 0, a: 1}, {_id: 1, a: 2}, {_id: 2, a: 3}];
assert.commandWorked(coll.insert(testData));
// Run the initial query and request to return a resume token. We will also request to return a
// recordId for the document as it will
// be used as a $_resumeAfter token later.
let res = assert.commandWorked(localDb.runCommand({
find: oplogCollName,
filter: {op: "i", "o.a": {$gte: 1}},
hint: {$natural: 1},
batchSize: 1,
showRecordId: true,
$_requestResumeToken: true
}));
assert.eq(1, res.cursor.firstBatch.length);
assert(res.cursor.firstBatch[0].hasOwnProperty("$recordId"));
assert.hasFields(res.cursor, ["postBatchResumeToken"]);
const firstResumeToken = res.cursor.postBatchResumeToken;
const firstOplogBatch = res.cursor.firstBatch;
res = assert.commandWorked(
localDb.runCommand({getMore: res.cursor.id, collection: oplogCollName, batchSize: 1}));
assert.eq(1, res.cursor.nextBatch.length);
assert(res.cursor.nextBatch[0].hasOwnProperty("$recordId"));
assert.hasFields(res.cursor, ["postBatchResumeToken"]);
const secondResumeToken = res.cursor.postBatchResumeToken;
const secondOplogBatch = res.cursor.nextBatch;
// Kill the cursor before attempting to resume.
assert.commandWorked(localDb.runCommand({killCursors: oplogCollName, cursors: [res.cursor.id]}));
// Try to resume the query from the saved resume token. This fails because '$_resumeAfter' expects a
// '$recordId' resume token, not a 'ts' resume token. This is appropriate since resuming on the
// oplog should occur by timestamp, not by record id.
assert.commandFailedWithCode(localDb.runCommand({
find: oplogCollName,
filter: {ts: {$gte: firstResumeToken.ts}},
hint: {$natural: 1},
batchSize: 1,
$_requestResumeToken: true,
$_resumeAfter: firstResumeToken
}),
ErrorCodes.BadValue);
// Now try to resume using a '$recordId' resume token, which doesn't make a ton of sense, but is
// still allowed.
res = assert.commandWorked(localDb.runCommand({
find: oplogCollName,
filter: {ts: {$gte: firstResumeToken.ts}},
hint: {$natural: 1},
batchSize: 1,
showRecordId: true,
$_requestResumeToken: true,
$_resumeAfter: {$recordId: firstOplogBatch[0].$recordId}
}));
assert.eq(1, res.cursor.firstBatch.length);
assert.eq(secondOplogBatch[0], res.cursor.firstBatch[0]);
// Kill the cursor before attempting to resume.
assert.commandWorked(localDb.runCommand({killCursors: oplogCollName, cursors: [res.cursor.id]}));
res = assert.commandWorked(localDb.runCommand({
find: oplogCollName,
filter: {ts: {$lte: secondResumeToken.ts}},
hint: {$natural: 1},
batchSize: 1,
showRecordId: true,
$_requestResumeToken: true,
$_resumeAfter: {$recordId: firstOplogBatch[0].$recordId}
}));
assert.eq(1, res.cursor.firstBatch.length);
assert.eq(secondOplogBatch[0], res.cursor.firstBatch[0]);
// Kill the cursor before attempting to resume.
assert.commandWorked(localDb.runCommand({killCursors: oplogCollName, cursors: [res.cursor.id]}));
res = assert.commandWorked(localDb.runCommand({
find: oplogCollName,
filter: {ts: {$eq: secondResumeToken.ts}},
hint: {$natural: 1},
batchSize: 1,
showRecordId: true,
$_requestResumeToken: true,
$_resumeAfter: {$recordId: firstOplogBatch[0].$recordId}
}));
assert.eq(1, res.cursor.firstBatch.length);
assert.eq(secondOplogBatch[0], res.cursor.firstBatch[0]);
// Kill the cursor before attempting to resume.
assert.commandWorked(localDb.runCommand({killCursors: oplogCollName, cursors: [res.cursor.id]}));
res = assert.commandWorked(localDb.runCommand({
find: oplogCollName,
filter: {ts: {$gt: firstResumeToken.ts, $lte: secondResumeToken.ts}},
hint: {$natural: 1},
batchSize: 1,
showRecordId: true,
$_requestResumeToken: true,
$_resumeAfter: {$recordId: firstOplogBatch[0].$recordId}
}));
assert.eq(1, res.cursor.firstBatch.length);
assert.eq(secondOplogBatch[0], res.cursor.firstBatch[0]);
// Kill the cursor before attempting to resume.
assert.commandWorked(localDb.runCommand({killCursors: oplogCollName, cursors: [res.cursor.id]}));
// Try to resume the query from a non-existent recordId and check that it fails to position the
// cursor to the record specified in the resume token.
assert.commandFailedWithCode(localDb.runCommand({
find: oplogCollName,
hint: {$natural: 1},
batchSize: 1,
$_requestResumeToken: true,
$_resumeAfter: {$recordId: NumberLong("50")}
}),
ErrorCodes.KeyNotFound);
// When we have a predicate on a 'ts' field of the oplog collection, we can build an optimized
// collection scan. Make sure we can run such optimized scans and get the result.
assert.eq(oplogColl.find({ts: {$gte: firstResumeToken.ts}}).itcount(), 3);
assert.eq(oplogColl.find({ts: {$gt: firstResumeToken.ts}}).itcount(), 2);
assert.gt(oplogColl.find({ts: {$lte: firstResumeToken.ts}}).itcount(), 1);
assert.gt(oplogColl.find({ts: {$lt: firstResumeToken.ts}}).itcount(), 1);
rst.stopSet();
})();
|
